Merge branch 'develop' into docs/purgeFiles

Closes https://github.com/kestra-io/kestra-ee/issues/4699

.github/ISSUE_TEMPLATE/bug.yml

@@ -2,6 +2,7 @@ name: Bug report
 description: Report a bug or unexpected behavior in the project
 
 labels: ["bug", "area/backend", "area/frontend"]
+type: Bug
 
 body:
   - type: markdown

.github/ISSUE_TEMPLATE/feature.yml

@@ -2,6 +2,7 @@ name: Feature request
 description: Suggest a new feature or improvement to enhance the project
 
 labels: ["enhancement", "area/backend", "area/frontend"]
+type: Feature
 
 body:
   - type: textarea

.github/dependabot.yml

@@ -26,7 +26,7 @@ updates:
     open-pull-requests-limit: 50
     labels: ["dependency-upgrade", "area/backend"]
     ignore:
-      # Ignore versions of Protobuf that are equal to or greater than 4.0.0 as Orc still uses 3
+      # Ignore versions of Protobuf >= 4.0.0 because Orc still uses version 3
       - dependency-name: "com.google.protobuf:*"
         versions: ["[4,)"]
 
@@ -44,68 +44,75 @@ updates:
       build:
         applies-to: version-updates
         patterns: ["@esbuild/*", "@rollup/*", "@swc/*"]
+
       types:
         applies-to: version-updates
         patterns: ["@types/*"]
+
       storybook:
         applies-to: version-updates
-        patterns: ["@storybook/*"]
+        patterns: ["storybook*", "@storybook/*", "eslint-plugin-storybook"]
+
       vitest:
         applies-to: version-updates
         patterns: ["vitest", "@vitest/*"]
-      patch:
+
+      major:
+        update-types: ["major"]
+        applies-to: version-updates
+        exclude-patterns: [
+            "@esbuild/*",
+            "@rollup/*",
+            "@swc/*",
+            "@types/*",
+            "storybook*",
+            "@storybook/*",
+            "eslint-plugin-storybook",
+            "vitest",
+            "@vitest/*",
+            # Temporary exclusion of these packages from major updates
+            "eslint-plugin-vue",
+          ]
+
+      minor:
+        update-types: ["minor"]
+        applies-to: version-updates
+        exclude-patterns: [
+            "@esbuild/*",
+            "@rollup/*",
+            "@swc/*",
+            "@types/*",
+            "storybook*",
+            "@storybook/*",
+            "eslint-plugin-storybook",
+            "vitest",
+            "@vitest/*",
+            # Temporary exclusion of these packages from minor updates
+            "moment-timezone",
+            "monaco-editor",
+          ]
+
+      patch:
+        update-types: ["patch"]
         applies-to: version-updates
-        patterns: ["*"]
         exclude-patterns:
           [
             "@esbuild/*",
             "@rollup/*",
             "@swc/*",
             "@types/*",
+            "storybook*",
             "@storybook/*",
-            "vitest",
-            "@vitest/*",
-          ]
-        update-types: ["patch"]
-      minor:
-        applies-to: version-updates
-        patterns: ["*"]
-        exclude-patterns: [
-            "@esbuild/*",
-            "@rollup/*",
-            "@swc/*",
-            "@types/*",
-            "@storybook/*",
-            "vitest",
-            "@vitest/*",
-            # Temporary exclusion of packages below from minor updates
-            "moment-timezone",
-            "monaco-editor",
-          ]
-        update-types: ["minor"]
-      major:
-        applies-to: version-updates
-        patterns: ["*"]
-        exclude-patterns: [
-            "@esbuild/*",
-            "@rollup/*",
-            "@swc/*",
-            "@types/*",
-            "@storybook/*",
-            "vitest",
-            "@vitest/*",
-            # Temporary exclusion of packages below from major updates
             "eslint-plugin-storybook",
-            "eslint-plugin-vue",
+            "vitest",
+            "@vitest/*",
           ]
-        update-types: ["major"]
-    ignore:
-      # Ignore updates to monaco-yaml, version is pinned to 5.3.1 due to patch-package script additions
-      - dependency-name: "monaco-yaml"
-        versions:
-          - ">=5.3.2"
 
-      # Ignore updates of version 1.x, as we're using the beta of 2.x (still in beta)
+    ignore:
+      # Ignore updates to monaco-yaml; version is pinned to 5.3.1 due to patch-package script additions
+      - dependency-name: "monaco-yaml"
+        versions: [">=5.3.2"]
+
+      # Ignore updates of version 1.x for vue-virtual-scroller, as the project uses the beta of 2.x
       - dependency-name: "vue-virtual-scroller"
-        versions:
-          - "1.x"
+        versions: ["1.x"]

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

@@ -20,7 +20,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 10
     steps:
-        - uses: actions/checkout@v5
+        - uses: actions/checkout@v6
           name: Checkout
           with:
             fetch-depth: 0

.github/workflows/codeql-analysis.yml

@@ -27,7 +27,7 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v5
+      uses: actions/checkout@v6
       with:
         # We must fetch at least the immediate parents so that if this is
         # a pull request then we can checkout the head.

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

@@ -33,7 +33,7 @@ jobs:
             exit 1;
           fi
       # Checkout
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v6
         with:
           fetch-depth: 0
           path: kestra

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

@@ -39,7 +39,7 @@ jobs:
 
       # Checkout
       - name: Checkout
-        uses: actions/checkout@v5
+        uses: actions/checkout@v6
         with:
           fetch-depth: 0
           token: ${{ secrets.GH_PERSONAL_TOKEN }}

.github/workflows/main-build.yml

@@ -28,7 +28,7 @@ jobs:
     steps:
       # Targeting develop branch from develop
       - name: Trigger EE Workflow (develop push, no payload)
-        uses: peter-evans/repository-dispatch@5fc4efd1a4797ddb68ffd0714a238564e4cc0e6f
+        uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697
         if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' }}
         with:
           token: ${{ secrets.GH_PERSONAL_TOKEN }}
@@ -64,6 +64,7 @@ jobs:
       DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
       DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
 
 
   publish-develop-maven:

.github/workflows/pull-request.yml

@@ -16,7 +16,7 @@ jobs:
     - name: Check EE repo for branch with same name
       if: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork == false }}
       id: check-ee-branch
-      uses: actions/github-script@v7
+      uses: actions/github-script@v8
       with:
         github-token: ${{ secrets.GH_PERSONAL_TOKEN }}
         script: |
@@ -40,7 +40,7 @@ jobs:
 
       # Targeting pull request (only if not from a fork and EE has no branch with same name)
     - name: Trigger EE Workflow (pull request, with payload)
-      uses: peter-evans/repository-dispatch@5fc4efd1a4797ddb68ffd0714a238564e4cc0e6f
+      uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697
       if: ${{ github.event_name == 'pull_request'
         && github.event.pull_request.number != ''
         && github.event.pull_request.head.repo.fork == false
@@ -50,7 +50,7 @@ jobs:
         repository: kestra-io/kestra-ee
         event-type: "oss-updated"
         client-payload: >-
-          {"commit_sha":"${{ github.sha }}","pr_repo":"${{ github.repository }}"}
+          {"commit_sha":"${{ github.event.pull_request.head.sha }}","pr_repo":"${{ github.repository }}"}
 
   file-changes:
     if: ${{ github.event.pull_request.draft == false }}

.github/workflows/release-docker.yml

@@ -32,3 +32,4 @@ jobs:
       DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
       DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}

.github/workflows/vulnerabilities-check.yml

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

build-and-start-e2e-tests.sh

@@ -29,8 +29,8 @@ start_time2=$(date +%s)
 
 echo "cd ./ui"
 cd ./ui
-echo "npm i"
-npm i
+echo "npm ci"
+npm ci
 
 echo 'sh ./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"'
 ./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"

build.gradle

@@ -7,7 +7,7 @@ buildscript {
     }
 
     dependencies {
-        classpath "net.e175.klaus:zip-prefixer:0.3.1"
+        classpath "net.e175.klaus:zip-prefixer:0.4.0"
     }
 }
 
@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "7.0.1.6134"
+    id "org.sonarqube" version "7.2.1.6560"
     id 'jacoco-report-aggregation'
 
     // helper
@@ -32,7 +32,7 @@ plugins {
 
     // release
     id 'net.researchgate.release' version '3.1.0'
-    id "com.gorylenko.gradle-git-properties" version "2.5.3"
+    id "com.gorylenko.gradle-git-properties" version "2.5.4"
     id 'signing'
     id "com.vanniktech.maven.publish" version "0.35.0"
 
@@ -223,13 +223,13 @@ subprojects {subProj ->
             t.environment 'ENV_TEST2', "Pass by env"
 
 
-            if (subProj.name == 'core' || subProj.name == 'jdbc-h2' || subProj.name == 'jdbc-mysql' || subProj.name == 'jdbc-postgres') {
-                // JUnit 5 parallel settings
-                t.systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
-                t.systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
-                t.systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
-                t.systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
-            }
+//            if (subProj.name == 'core' || subProj.name == 'jdbc-h2' || subProj.name == 'jdbc-mysql' || subProj.name == 'jdbc-postgres') {
+//                // JUnit 5 parallel settings
+//                t.systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
+//                t.systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
+//                t.systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
+//                t.systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
+//            }
         }
 
         tasks.register('flakyTest', Test) { Test t ->
@@ -331,7 +331,7 @@ subprojects {
         }
 
         dependencies {
-            agent "org.aspectj:aspectjweaver:1.9.25"
+            agent "org.aspectj:aspectjweaver:1.9.25.1"
         }
 
         test {

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

@@ -93,7 +93,7 @@ public class App implements Callable<Integer> {
         try {
              exitCode = new CommandLine(cls, new MicronautFactory(applicationContext)).execute(args);
         } catch (CommandLine.InitializationException e){
-            System.err.println("Could not initialize picoli ComandLine, err: " + e.getMessage());
+            System.err.println("Could not initialize picocli CommandLine, err: " + e.getMessage());
             e.printStackTrace();
             exitCode = 1;
         }

cli/src/main/java/io/kestra/cli/commands/migrations/metadata/MetadataMigrationCommand.java

@@ -10,7 +10,8 @@ import picocli.CommandLine;
     description = "populate metadata for entities",
     subcommands = {
         KvMetadataMigrationCommand.class,
-        SecretsMetadataMigrationCommand.class
+        SecretsMetadataMigrationCommand.class,
+        NsFilesMetadataMigrationCommand.class
     }
 )
 @Slf4j

cli/src/main/java/io/kestra/cli/commands/migrations/metadata/MetadataMigrationService.java

@@ -1,47 +1,51 @@
 package io.kestra.cli.commands.migrations.metadata;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.models.kv.PersistedKvMetadata;
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.KvMetadataRepositoryInterface;
+import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
 import io.kestra.core.storages.FileAttributes;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.storages.kv.InternalKVStore;
 import io.kestra.core.storages.kv.KVEntry;
 import io.kestra.core.tenant.TenantService;
-import jakarta.inject.Inject;
+import io.kestra.core.utils.NamespaceUtils;
 import jakarta.inject.Singleton;
+import lombok.AllArgsConstructor;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.file.NoSuchFileException;
 import java.time.Instant;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
+import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static io.kestra.core.utils.Rethrow.throwConsumer;
 import static io.kestra.core.utils.Rethrow.throwFunction;
 
 @Singleton
+@AllArgsConstructor
 public class MetadataMigrationService {
-    @Inject
-    private TenantService tenantService;
+    protected FlowRepositoryInterface flowRepository;
+    protected TenantService tenantService;
+    protected KvMetadataRepositoryInterface kvMetadataRepository;
+    protected NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository;
+    protected StorageInterface storageInterface;
+    protected NamespaceUtils namespaceUtils;
 
-    @Inject
-    private FlowRepositoryInterface flowRepository;
-
-    @Inject
-    private KvMetadataRepositoryInterface kvMetadataRepository;
-
-    @Inject
-    private StorageInterface storageInterface;
-
-    protected Map<String, List<String>> namespacesPerTenant() {
+    @VisibleForTesting
+    public Map<String, List<String>> namespacesPerTenant() {
         String tenantId = tenantService.resolveTenant();
-        return Map.of(tenantId, flowRepository.findDistinctNamespace(tenantId));
+        return Map.of(tenantId, Stream.concat(
+            Stream.of(namespaceUtils.getSystemFlowNamespace()),
+            flowRepository.findDistinctNamespace(tenantId).stream()
+        ).map(NamespaceUtils::asTree).flatMap(Collection::stream).distinct().toList());
     }
 
     public void kvMigration() throws IOException {
@@ -49,7 +53,9 @@ public class MetadataMigrationService {
             .flatMap(namespacesForTenant -> namespacesForTenant.getValue().stream().map(namespace -> Map.entry(namespacesForTenant.getKey(), namespace)))
             .flatMap(throwFunction(namespaceForTenant -> {
                 InternalKVStore kvStore = new InternalKVStore(namespaceForTenant.getKey(), namespaceForTenant.getValue(), storageInterface, kvMetadataRepository);
-                List<FileAttributes> list = listAllFromStorage(storageInterface, namespaceForTenant.getKey(), namespaceForTenant.getValue());
+                List<FileAttributes> list = listAllFromStorage(storageInterface, StorageContext::kvPrefix, namespaceForTenant.getKey(), namespaceForTenant.getValue()).stream()
+                    .map(PathAndAttributes::attributes)
+                    .toList();
                 Map<Boolean, List<KVEntry>> entriesByIsExpired = list.stream()
                     .map(throwFunction(fileAttributes -> KVEntry.from(namespaceForTenant.getValue(), fileAttributes)))
                     .collect(Collectors.partitioningBy(kvEntry -> Optional.ofNullable(kvEntry.expirationDate()).map(expirationDate -> Instant.now().isAfter(expirationDate)).orElse(false)));
@@ -75,15 +81,39 @@ public class MetadataMigrationService {
             }));
     }
 
+    public void nsFilesMigration() throws IOException {
+        this.namespacesPerTenant().entrySet().stream()
+            .flatMap(namespacesForTenant -> namespacesForTenant.getValue().stream().map(namespace -> Map.entry(namespacesForTenant.getKey(), namespace)))
+            .flatMap(throwFunction(namespaceForTenant -> {
+                List<PathAndAttributes> list = listAllFromStorage(storageInterface, StorageContext::namespaceFilePrefix, namespaceForTenant.getKey(), namespaceForTenant.getValue());
+                return list.stream()
+                    .map(pathAndAttributes -> NamespaceFileMetadata.of(namespaceForTenant.getKey(), namespaceForTenant.getValue(), pathAndAttributes.path(), pathAndAttributes.attributes()));
+            }))
+            .forEach(throwConsumer(nsFileMetadata -> {
+                if (namespaceFileMetadataRepository.findByPath(nsFileMetadata.getTenantId(), nsFileMetadata.getNamespace(), nsFileMetadata.getPath()).isEmpty()) {
+                    namespaceFileMetadataRepository.save(nsFileMetadata);
+                }
+            }));
+    }
+
     public void secretMigration() throws Exception {
         throw new UnsupportedOperationException("Secret migration is not needed in the OSS version");
     }
 
-    private static List<FileAttributes> listAllFromStorage(StorageInterface storage, String tenant, String namespace) throws IOException {
+    private static List<PathAndAttributes> listAllFromStorage(StorageInterface storage, Function<String, String> prefixFunction, String tenant, String namespace) throws IOException {
         try {
-            return storage.list(tenant, namespace, URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.kvPrefix(namespace)));
-        } catch (FileNotFoundException e) {
+            String prefix = prefixFunction.apply(namespace);
+            if (!storage.exists(tenant, namespace, URI.create(StorageContext.KESTRA_PROTOCOL + prefix))) {
+                return Collections.emptyList();
+            }
+
+            return storage.allByPrefix(tenant, namespace, URI.create(StorageContext.KESTRA_PROTOCOL + prefix + "/"), true).stream()
+                .map(throwFunction(uri -> new PathAndAttributes(uri.getPath().substring(prefix.length()), storage.getAttributes(tenant, namespace, uri))))
+                .toList();
+        } catch (FileNotFoundException | NoSuchFileException e) {
             return Collections.emptyList();
         }
     }
+
+    public record PathAndAttributes(String path, FileAttributes attributes) {}
 }

cli/src/main/java/io/kestra/cli/commands/migrations/metadata/NsFilesMetadataMigrationCommand.java

@@ -0,0 +1,31 @@
+package io.kestra.cli.commands.migrations.metadata;
+
+import io.kestra.cli.AbstractCommand;
+import jakarta.inject.Inject;
+import jakarta.inject.Provider;
+import lombok.extern.slf4j.Slf4j;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+    name = "nsfiles",
+    description = "populate metadata for Namespace Files"
+)
+@Slf4j
+public class NsFilesMetadataMigrationCommand extends AbstractCommand {
+    @Inject
+    private Provider<MetadataMigrationService> metadataMigrationServiceProvider;
+
+    @Override
+    public Integer call() throws Exception {
+        super.call();
+        try {
+            metadataMigrationServiceProvider.get().nsFilesMigration();
+        } catch (Exception e) {
+            System.err.println("❌ Namespace Files Metadata migration failed: " + e.getMessage());
+            e.printStackTrace();
+            return 1;
+        }
+        System.out.println("✅ Namespace Files Metadata migration complete.");
+        return 0;
+    }
+}

cli/src/main/java/io/kestra/cli/commands/sys/statestore/StateStoreMigrateCommand.java

@@ -57,7 +57,7 @@ public class StateStoreMigrateCommand extends AbstractCommand {
             String taskRunValue = statesUriPart.length > 2 ? statesUriPart[1] : null;
             String stateSubName = statesUriPart[statesUriPart.length - 1];
             boolean flowScoped = flowQualifierWithStateQualifiers[0].endsWith("/" + flow.getId());
-            StateStore stateStore = new StateStore(runContext(runContextFactory, flow), false);
+            StateStore stateStore = new StateStore(runContextFactory.of(flow, Map.of()), false);
 
             try (InputStream is = storageInterface.get(flow.getTenantId(), flow.getNamespace(), stateStoreFileUri)) {
                 stateStore.putState(flowScoped, stateName, stateSubName, taskRunValue, is.readAllBytes());
@@ -70,12 +70,4 @@ public class StateStoreMigrateCommand extends AbstractCommand {
         stdOut("Successfully ran the state-store migration.");
         return 0;
     }
-
-    private RunContext runContext(RunContextFactory runContextFactory, Flow flow) {
-        Map<String, String> flowVariables = new HashMap<>();
-        flowVariables.put("tenantId", flow.getTenantId());
-        flowVariables.put("id", flow.getId());
-        flowVariables.put("namespace", flow.getNamespace());
-        return runContextFactory.of(flow, Map.of("flow", flowVariables));
-    }
 }

cli/src/test/java/io/kestra/cli/commands/migrations/metadata/MetadataMigrationServiceTest.java

@@ -0,0 +1,57 @@
+package io.kestra.cli.commands.migrations.metadata;
+
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.tenant.TenantService;
+import io.kestra.core.utils.NamespaceUtils;
+import io.kestra.core.utils.TestsUtils;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class MetadataMigrationServiceTest<T extends MetadataMigrationService> {
+    private static final String TENANT_ID = TestsUtils.randomTenant();
+
+    protected static final String SYSTEM_NAMESPACE = "my.system.namespace";
+
+    @Test
+    void namespacesPerTenant() {
+        Map<String, List<String>> expected = getNamespacesPerTenant();
+        Map<String, List<String>> result = metadataMigrationService(
+            expected
+        ).namespacesPerTenant();
+
+        assertThat(result).hasSize(expected.size());
+        expected.forEach((tenantId, namespaces) -> {
+            assertThat(result.get(tenantId)).containsExactlyInAnyOrderElementsOf(
+                Stream.concat(
+                    Stream.of(SYSTEM_NAMESPACE),
+                    namespaces.stream()
+                ).map(NamespaceUtils::asTree).flatMap(Collection::stream).distinct().toList()
+            );
+        });
+    }
+
+    protected Map<String, List<String>> getNamespacesPerTenant() {
+        return Map.of(TENANT_ID, List.of("my.first.namespace", "my.second.namespace", "another.namespace"));
+    }
+
+    protected T metadataMigrationService(Map<String, List<String>> namespacesPerTenant) {
+        FlowRepositoryInterface mockedFlowRepository = Mockito.mock(FlowRepositoryInterface.class);
+        Mockito.doAnswer((params) -> namespacesPerTenant.get(params.getArgument(0).toString())).when(mockedFlowRepository).findDistinctNamespace(Mockito.anyString());
+        NamespaceUtils namespaceUtils = Mockito.mock(NamespaceUtils.class);
+        Mockito.when(namespaceUtils.getSystemFlowNamespace()).thenReturn(SYSTEM_NAMESPACE);
+        //noinspection unchecked
+        return ((T) new MetadataMigrationService(mockedFlowRepository, new TenantService() {
+            @Override
+            public String resolveTenant() {
+                return TENANT_ID;
+            }
+        }, null, null, null, namespaceUtils));
+    }
+}

cli/src/test/java/io/kestra/cli/commands/migrations/metadata/NsFilesMetadataMigrationCommandTest.java

@@ -0,0 +1,175 @@
+package io.kestra.cli.commands.migrations.metadata;
+
+import io.kestra.cli.App;
+import io.kestra.core.exceptions.ResourceExpiredException;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.GenericFlow;
+import io.kestra.core.models.kv.PersistedKvMetadata;
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.KvMetadataRepositoryInterface;
+import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
+import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.storages.*;
+import io.kestra.core.storages.kv.*;
+import io.kestra.core.tenant.TenantService;
+import io.kestra.core.utils.TestsUtils;
+import io.kestra.plugin.core.log.Log;
+import io.micronaut.configuration.picocli.PicocliRunner;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.env.Environment;
+import io.micronaut.core.annotation.NonNull;
+import org.junit.jupiter.api.Test;
+
+import java.io.*;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class NsFilesMetadataMigrationCommandTest {
+    @Test
+    void run() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+        ByteArrayOutputStream err = new ByteArrayOutputStream();
+        System.setErr(new PrintStream(err));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+            /* Initial setup:
+            * - namespace 1: my/path, value
+            * - namespace 1: another/path
+            * - namespace 2: yet/another/path
+            * - Nothing in database */
+            String namespace = TestsUtils.randomNamespace();
+            String path = "/my/path";
+            StorageInterface storage = ctx.getBean(StorageInterface.class);
+            String value = "someValue";
+            putOldNsFile(storage, namespace, path, value);
+
+            String anotherPath = "/another/path";
+            String anotherValue = "anotherValue";
+            putOldNsFile(storage, namespace, anotherPath, anotherValue);
+
+            String anotherNamespace = TestsUtils.randomNamespace();
+            String yetAnotherPath = "/yet/another/path";
+            String yetAnotherValue = "yetAnotherValue";
+            putOldNsFile(storage, anotherNamespace, yetAnotherPath, yetAnotherValue);
+
+            NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository = ctx.getBean(NamespaceFileMetadataRepositoryInterface.class);
+            String tenantId = TenantService.MAIN_TENANT;
+            assertThat(namespaceFileMetadataRepository.findByPath(tenantId, namespace, path).isPresent()).isFalse();
+
+            /* Expected outcome from the migration command:
+            * - no namespace files has been migrated because no flow exist in the namespace so they are not picked up because we don't know they exist */
+            String[] nsFilesMetadataMigrationCommand = {
+                "migrate", "metadata", "nsfiles"
+            };
+            PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
+
+
+            assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
+            // Still it's not in the metadata repository because no flow exist to find that namespace file
+            assertThat(namespaceFileMetadataRepository.findByPath(tenantId, namespace, path).isPresent()).isFalse();
+            assertThat(namespaceFileMetadataRepository.findByPath(tenantId, namespace, anotherPath).isPresent()).isFalse();
+            assertThat(namespaceFileMetadataRepository.findByPath(tenantId, anotherNamespace, yetAnotherPath).isPresent()).isFalse();
+
+            // A flow is created from namespace 1, so the namespace files in this namespace should be migrated
+            FlowRepositoryInterface flowRepository = ctx.getBean(FlowRepositoryInterface.class);
+            flowRepository.create(GenericFlow.of(Flow.builder()
+                .tenantId(tenantId)
+                .id("a-flow")
+                .namespace(namespace)
+                .tasks(List.of(Log.builder().id("log").type(Log.class.getName()).message("logging").build()))
+                .build()));
+
+            /* We run the migration again:
+            * - namespace 1 my/path file is seen and metadata is migrated to database
+            * - namespace 1 another/path file is seen and metadata is migrated to database
+            * - namespace 2 yet/another/path is not seen because no flow exist in this namespace */
+            out.reset();
+            PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
+
+            assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
+            Optional<NamespaceFileMetadata> foundNsFile = namespaceFileMetadataRepository.findByPath(tenantId, namespace, path);
+            assertThat(foundNsFile.isPresent()).isTrue();
+            assertThat(foundNsFile.get().getVersion()).isEqualTo(1);
+            assertThat(foundNsFile.get().getSize()).isEqualTo(value.length());
+
+            Optional<NamespaceFileMetadata> anotherFoundNsFile = namespaceFileMetadataRepository.findByPath(tenantId, namespace, anotherPath);
+            assertThat(anotherFoundNsFile.isPresent()).isTrue();
+            assertThat(anotherFoundNsFile.get().getVersion()).isEqualTo(1);
+            assertThat(anotherFoundNsFile.get().getSize()).isEqualTo(anotherValue.length());
+
+            NamespaceFactory namespaceFactory = ctx.getBean(NamespaceFactory.class);
+            Namespace namespaceStorage = namespaceFactory.of(tenantId, namespace, storage);
+            FileAttributes nsFileRawMetadata = namespaceStorage.getFileMetadata(Path.of(path));
+            assertThat(nsFileRawMetadata.getSize()).isEqualTo(value.length());
+            assertThat(new String(namespaceStorage.getFileContent(Path.of(path)).readAllBytes())).isEqualTo(value);
+
+            FileAttributes anotherNsFileRawMetadata = namespaceStorage.getFileMetadata(Path.of(anotherPath));
+            assertThat(anotherNsFileRawMetadata.getSize()).isEqualTo(anotherValue.length());
+            assertThat(new String(namespaceStorage.getFileContent(Path.of(anotherPath)).readAllBytes())).isEqualTo(anotherValue);
+
+            assertThat(namespaceFileMetadataRepository.findByPath(tenantId, anotherNamespace, yetAnotherPath).isPresent()).isFalse();
+            assertThatThrownBy(() -> namespaceStorage.getFileMetadata(Path.of(yetAnotherPath))).isInstanceOf(FileNotFoundException.class);
+
+            /* We run one last time the migration without any change to verify that we don't resave an existing metadata.
+            * It covers the case where user didn't perform the migrate command yet but they played and added some KV from the UI (so those ones will already be in metadata database). */
+            out.reset();
+            PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
+
+            assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
+            foundNsFile = namespaceFileMetadataRepository.findByPath(tenantId, namespace, path);
+            assertThat(foundNsFile.get().getVersion()).isEqualTo(1);
+        }
+    }
+
+    @Test
+    void namespaceWithoutNsFile() {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+        ByteArrayOutputStream err = new ByteArrayOutputStream();
+        System.setErr(new PrintStream(err));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+            String tenantId = TenantService.MAIN_TENANT;
+            String namespace = TestsUtils.randomNamespace();
+
+            // A flow is created from namespace 1, so the namespace files in this namespace should be migrated
+            FlowRepositoryInterface flowRepository = ctx.getBean(FlowRepositoryInterface.class);
+            flowRepository.create(GenericFlow.of(Flow.builder()
+                .tenantId(tenantId)
+                .id("a-flow")
+                .namespace(namespace)
+                .tasks(List.of(Log.builder().id("log").type(Log.class.getName()).message("logging").build()))
+                .build()));
+
+            String[] nsFilesMetadataMigrationCommand = {
+                "migrate", "metadata", "nsfiles"
+            };
+            PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
+
+            assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
+            assertThat(err.toString()).doesNotContain("java.nio.file.NoSuchFileException");
+        }
+    }
+
+    private static void putOldNsFile(StorageInterface storage, String namespace, String path, String value) throws IOException {
+        URI nsFileStorageUri = getNsFileStorageUri(namespace, path);
+        storage.put(TenantService.MAIN_TENANT, namespace, nsFileStorageUri, new StorageObject(
+            null,
+            new ByteArrayInputStream(value.getBytes(StandardCharsets.UTF_8))
+        ));
+    }
+
+    private static @NonNull URI getNsFileStorageUri(String namespace, String path) {
+        return URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.namespaceFilePrefix(namespace) + path);
+    }
+}

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

@@ -55,11 +55,7 @@ class StateStoreMigrateCommandTest {
             );
             assertThat(storage.exists(tenantId, flow.getNamespace(), oldStateStoreUri)).isTrue();
 
-            RunContext runContext = ctx.getBean(RunContextFactory.class).of(flow, Map.of("flow", Map.of(
-                "tenantId", tenantId,
-                "id", flow.getId(),
-                "namespace", flow.getNamespace()
-            )));
+            RunContext runContext = ctx.getBean(RunContextFactory.class).of(flow, Map.of());
             StateStore stateStore = new StateStore(runContext, true);
             Assertions.assertThrows(MigrationRequiredException.class, () -> stateStore.getState(true, "my-state", "sub-name", "my-taskrun-value"));
 

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

@@ -19,7 +19,6 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import org.junitpioneer.jupiter.RetryingTest;
 
 import static io.kestra.core.utils.Rethrow.throwRunnable;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -59,7 +58,7 @@ class FileChangedEventListenerTest {
     }
 
     @FlakyTest
-    @RetryingTest(2)
+    @Test
     void test() throws IOException, TimeoutException {
         var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getSimpleName(), "test");
         // remove the flow if it already exists
@@ -98,7 +97,7 @@ class FileChangedEventListenerTest {
     }
 
     @FlakyTest
-    @RetryingTest(2)
+    @Test
     void testWithPluginDefault() throws IOException, TimeoutException {
         var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getName(), "testWithPluginDefault");
         // remove the flow if it already exists
@@ -138,4 +137,4 @@ class FileChangedEventListenerTest {
             Duration.ofSeconds(10)
         );
     }
-}
+}

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

@@ -21,6 +21,7 @@ kestra:
   server:
     liveness:
       enabled: false
+    termination-grace-period: 5s
 micronaut:
   http:
     services:

core/build.gradle

@@ -82,8 +82,8 @@ dependencies {
     testImplementation "io.micronaut:micronaut-http-server-netty"
     testImplementation "io.micronaut:micronaut-management"
 
-    testImplementation "org.testcontainers:testcontainers:1.21.3"
-    testImplementation "org.testcontainers:junit-jupiter:1.21.3"
+    testImplementation "org.testcontainers:testcontainers:1.21.4"
+    testImplementation "org.testcontainers:junit-jupiter:1.21.4"
     testImplementation "org.bouncycastle:bcpkix-jdk18on"
 
     testImplementation "org.wiremock:wiremock-jetty12"

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

@@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -84,6 +85,11 @@ public abstract class KestraContext {
 
     public abstract StorageInterface getStorageInterface();
 
+    /**
+     * Returns the Micronaut active environments.
+     */
+    public abstract Set<String> getEnvironments();
+
     /**
      * Shutdowns the Kestra application.
      */
@@ -182,5 +188,10 @@ public abstract class KestraContext {
             // Lazy init of the PluginRegistry.
             return this.applicationContext.getBean(StorageInterface.class);
         }
+
+        @Override
+        public Set<String> getEnvironments() {
+            return this.applicationContext.getEnvironment().getActiveNames();
+        }
     }
 }

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

@@ -3,6 +3,7 @@ package io.kestra.core.docs;
 import io.kestra.core.models.annotations.PluginSubGroup;
 import io.kestra.core.plugins.RegisteredPlugin;
 import io.micronaut.core.annotation.Nullable;
+import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
@@ -117,10 +118,17 @@ public class Plugin {
             .filter(not(io.kestra.core.models.Plugin::isInternal))
             .filter(clazzFilter)
             .filter(c -> !c.getName().startsWith("org.kestra."))
-            .map(c -> new PluginElementMetadata(c.getName(), io.kestra.core.models.Plugin.isDeprecated(c) ? true : null))
+            .map(c -> {
+                Schema schema = c.getAnnotation(Schema.class);
+
+                var title = Optional.ofNullable(schema).map(Schema::title).filter(t -> !t.isEmpty()).orElse(null);
+                var description = Optional.ofNullable(schema).map(Schema::description).filter(d -> !d.isEmpty()).orElse(null);
+                var deprecated = io.kestra.core.models.Plugin.isDeprecated(c) ? true : null;
+
+                return new PluginElementMetadata(c.getName(), deprecated, title, description);
+            })
             .toList();
     }
 
-    public record PluginElementMetadata(String cls, Boolean deprecated) {
-    }
+    public record PluginElementMetadata(String cls, Boolean deprecated, String title, String description) {}
 }

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

@@ -0,0 +1,23 @@
+package io.kestra.core.exceptions;
+
+/**
+ * Exception that can be thrown when a Flow is not found.
+ */
+public class FlowNotFoundException extends NotFoundException {
+
+    /**
+     * Creates a new {@link FlowNotFoundException} instance.
+     */
+    public FlowNotFoundException() {
+        super();
+    }
+
+    /**
+     * Creates a new {@link NotFoundException} instance.
+     *
+     * @param message the error message.
+     */
+    public FlowNotFoundException(final String message) {
+        super(message);
+    }
+}

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

@@ -0,0 +1,15 @@
+package io.kestra.core.exceptions;
+
+import java.io.Serial;
+
+public class ResourceAccessDeniedException extends KestraRuntimeException {
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    public ResourceAccessDeniedException() {
+    }
+
+    public ResourceAccessDeniedException(String message) {
+        super(message);
+    }
+}

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

@@ -7,7 +7,6 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
 import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipInputStream;
@@ -65,7 +64,7 @@ public interface HasSource {
 
             if (isYAML(fileName)) {
                 byte[] bytes = inputStream.readAllBytes();
-                List<String> sources = List.of(new String(bytes).split("---"));
+                List<String> sources = List.of(new String(bytes).split("(?m)^---\\s*$"));
                 for (int i = 0; i < sources.size(); i++) {
                     String source = sources.get(i);
                     reader.accept(source, String.valueOf(i));

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

@@ -4,13 +4,16 @@ import io.kestra.core.utils.MapUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.annotation.Nullable;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.Pattern;
 
 import java.util.*;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 @Schema(description = "A key/value pair that can be attached to a Flow or Execution. Labels are often used to organize and categorize objects.")
-public record Label(@NotEmpty String key, @NotEmpty String value) {
+public record Label(
+        @NotEmpty @Pattern(regexp = "^[\\p{Ll}][\\p{L}0-9._-]*$", message = "Invalid label key. A valid key contains only lowercase letters numbers hyphens (-) underscores (_) or periods (.) and must begin with a lowercase letter.") String key,
+        @NotEmpty String value) {
     public static final String SYSTEM_PREFIX = "system.";
 
     // system labels
@@ -23,6 +26,7 @@ public record Label(@NotEmpty String key, @NotEmpty String value) {
     public static final String REPLAYED = SYSTEM_PREFIX + "replayed";
     public static final String SIMULATED_EXECUTION = SYSTEM_PREFIX + "simulatedExecution";
     public static final String TEST = SYSTEM_PREFIX + "test";
+    public static final String FROM = SYSTEM_PREFIX + "from";
 
     /**
      * Static helper method for converting a list of labels to a nested map.

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

@@ -94,7 +94,7 @@ public record QueryFilter(
         KIND("kind") {
             @Override
             public List<Op> supportedOp() {
-                return List.of(Op.EQUALS,Op.NOT_EQUALS);
+                return List.of(Op.EQUALS,Op.NOT_EQUALS, Op.IN, Op.NOT_IN);
             }
         },
         LABELS("labels") {
@@ -106,7 +106,7 @@ public record QueryFilter(
         FLOW_ID("flowId") {
             @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, Op.NOT_IN, Op.PREFIX);
             }
         },
         UPDATED("updated") {
@@ -180,6 +180,24 @@ public record QueryFilter(
             public List<Op> supportedOp() {
                 return List.of(Op.EQUALS, Op.NOT_EQUALS);
             }
+        },
+        PATH("path") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.IN);
+            }
+        },
+        PARENT_PATH("parentPath") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.STARTS_WITH);
+            }
+        },
+        VERSION("version") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS);
+            }
         };
 
         private static final Map<String, Field> BY_VALUE = Arrays.stream(values())
@@ -208,7 +226,7 @@ public record QueryFilter(
         FLOW {
             @Override
             public List<Field> supportedField() {
-                return List.of(Field.LABELS, Field.NAMESPACE, Field.QUERY, Field.SCOPE);
+                return List.of(Field.LABELS, Field.NAMESPACE, Field.QUERY, Field.SCOPE, Field.FLOW_ID);
             }
         },
         NAMESPACE {
@@ -223,7 +241,7 @@ public record QueryFilter(
                 return List.of(
                     Field.QUERY, Field.SCOPE, Field.FLOW_ID, Field.START_DATE, Field.END_DATE,
                     Field.STATE, Field.LABELS, Field.TRIGGER_EXECUTION_ID, Field.CHILD_FILTER,
-                    Field.NAMESPACE,Field.KIND
+                    Field.NAMESPACE, Field.KIND
                 );
             }
         },
@@ -275,6 +293,19 @@ public record QueryFilter(
                     Field.UPDATED
                 );
             }
+        },
+        NAMESPACE_FILE_METADATA {
+            @Override
+            public List<Field> supportedField() {
+                return List.of(
+                    Field.QUERY,
+                    Field.NAMESPACE,
+                    Field.PATH,
+                    Field.PARENT_PATH,
+                    Field.VERSION,
+                    Field.UPDATED
+                );
+            }
         };
 
         public abstract List<Field> supportedField();

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

@@ -16,6 +16,7 @@ import jakarta.validation.constraints.NotNull;
 public class Setting {
     public static final String INSTANCE_UUID = "instance.uuid";
     public static final String INSTANCE_VERSION = "instance.version";
+    public static final String INSTANCE_EDITION = "instance.edition";
 
     @NotNull
     private String key;

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

@@ -658,18 +658,20 @@ public class Execution implements DeletedInterface, TenantInterface {
     public boolean hasFailedNoRetry(List<ResolvedTask> resolvedTasks, TaskRun parentTaskRun) {
         return this.findTaskRunByTasks(resolvedTasks, parentTaskRun)
             .stream()
-            .anyMatch(taskRun -> {
-                ResolvedTask resolvedTask = resolvedTasks.stream()
-                    .filter(t -> t.getTask().getId().equals(taskRun.getTaskId())).findFirst()
-                    .orElse(null);
-                if (resolvedTask == null) {
-                    log.warn("Can't find task for taskRun '{}' in parentTaskRun '{}'",
-                        taskRun.getId(), parentTaskRun.getId());
-                    return false;
-                }
-                return !taskRun.shouldBeRetried(resolvedTask.getTask().getRetry())
-                    && taskRun.getState().isFailed();
-            });
+            // NOTE: we check on isFailed first to avoid the costly shouldBeRetried() method
+            .anyMatch(taskRun -> taskRun.getState().isFailed() && shouldNotBeRetried(resolvedTasks, parentTaskRun, taskRun));
+    }
+
+    private static boolean shouldNotBeRetried(List<ResolvedTask> resolvedTasks, TaskRun parentTaskRun, TaskRun taskRun) {
+        ResolvedTask resolvedTask = resolvedTasks.stream()
+            .filter(t -> t.getTask().getId().equals(taskRun.getTaskId())).findFirst()
+            .orElse(null);
+        if (resolvedTask == null) {
+            log.warn("Can't find task for taskRun '{}' in parentTaskRun '{}'",
+                taskRun.getId(), parentTaskRun.getId());
+            return false;
+        }
+        return !taskRun.shouldBeRetried(resolvedTask.getTask().getRetry());
     }
 
     public boolean hasCreated() {

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

@@ -1,15 +1,16 @@
 package io.kestra.core.models.executions;
 
-import io.micronaut.core.annotation.Introspected;
-import lombok.Builder;
-import lombok.Value;
 import io.kestra.core.models.tasks.Output;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.micronaut.core.annotation.Introspected;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
+import lombok.Builder;
+import lombok.Value;
 
 import java.net.URI;
 import java.util.Collections;
 import java.util.Map;
-import jakarta.validation.constraints.NotNull;
 
 @Value
 @Builder
@@ -21,6 +22,7 @@ public class ExecutionTrigger {
     @NotNull
     String type;
 
+    @Schema(type = "object", additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
     Map<String, Object> variables;
 
     URI logFile;

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

@@ -314,4 +314,11 @@ public class TaskRun implements TenantInterface {
             .build();
     }
 
+    public TaskRun addAttempt(TaskRunAttempt attempt) {
+        if (this.attempts == null) {
+            this.attempts = new ArrayList<>();
+        }
+        this.attempts.add(attempt);
+        return this;
+    }
 }

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

@@ -24,4 +24,8 @@ public class Concurrency {
     public enum Behavior {
         QUEUE, CANCEL, FAIL;
     }
+
+    public static boolean possibleTransitions(State.Type type) {
+        return type.equals(State.Type.CANCELLED) || type.equals(State.Type.FAILED);
+    }
 }

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

@@ -1,6 +1,5 @@
 package io.kestra.core.models.flows;
 
-import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -11,6 +10,7 @@ import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.HasUID;
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.flows.check.Check;
 import io.kestra.core.models.flows.sla.SLA;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
@@ -130,6 +130,14 @@ public class Flow extends AbstractFlow implements HasUID {
     @PluginProperty
     List<SLA> sla;
 
+    @Schema(
+        title = "Conditions evaluated before the flow is executed.",
+        description = "A list of conditions that are evaluated before the flow is executed.  If no checks are defined, the flow executes normally."
+    )
+    @Valid
+    @PluginProperty
+    List<Check> checks;
+
     public Stream<String> allTypes() {
         return Stream.of(
                 Optional.ofNullable(triggers).orElse(Collections.emptyList()).stream().map(AbstractTrigger::getType),
@@ -346,7 +354,7 @@ public class Flow extends AbstractFlow implements HasUID {
      * To be conservative a flow MUST not return any source.
      */
     @Override
-    @JsonIgnore
+    @Schema(hidden = true)
     public String getSource() {
         return null;
     }

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

@@ -1,14 +1,12 @@
 package io.kestra.core.models.flows;
 
-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 io.swagger.v3.oas.annotations.media.Schema;
 
-import java.util.Objects;
-import java.util.regex.Pattern;
 
 @SuperBuilder(toBuilder = true)
 @Getter
@@ -43,11 +41,12 @@ public class FlowWithSource extends Flow {
             .concurrency(this.concurrency)
             .retry(this.retry)
             .sla(this.sla)
+            .checks(this.checks)
             .build();
     }
 
     @Override
-    @JsonIgnore(value = false)
+    @Schema(hidden = false)
     public String getSource() {
         return this.source;
     }
@@ -85,6 +84,7 @@ public class FlowWithSource extends Flow {
             .concurrency(flow.concurrency)
             .retry(flow.retry)
             .sla(flow.sla)
+            .checks(flow.checks)
             .build();
     }
 }

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

@@ -84,12 +84,24 @@ public class State {
         );
     }
 
+    /**
+     * non-terminated execution duration is hard to provide in SQL, so we set it to null when endDate is empty
+     */
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)
-    public Duration getDuration() {
-        return Duration.between(
-            this.histories.getFirst().getDate(),
-            this.histories.size() > 1 ? this.histories.get(this.histories.size() - 1).getDate() : Instant.now()
-        );
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public Optional<Duration> getDuration() {
+        if (this.getEndDate().isPresent()) {
+            return Optional.of(Duration.between(this.getStartDate(), this.getEndDate().get()));
+        } else {
+            return Optional.empty();
+        }
+    }
+
+    /**
+     * @return either the Duration persisted in database, or calculate it on the fly for non-terminated executions
+     */
+    public Duration getDurationOrComputeIt() {
+        return this.getDuration().orElseGet(() -> Duration.between(this.getStartDate(), Instant.now()));
     }
 
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)
@@ -109,7 +121,7 @@ public class State {
 
     public String humanDuration() {
         try {
-            return DurationFormatUtils.formatDurationHMS(getDuration().toMillis());
+            return DurationFormatUtils.formatDurationHMS(getDurationOrComputeIt().toMillis());
         } catch (Throwable e) {
             return getDuration().toString();
         }
@@ -255,6 +267,10 @@ public class State {
             return this == Type.RUNNING || this == Type.KILLING;
         }
 
+        public boolean onlyRunning() {
+            return this == Type.RUNNING;
+        }
+
         public boolean isFailed() {
             return this == Type.FAILED;
         }

core/src/main/java/io/kestra/core/models/flows/check/Check.java

@@ -0,0 +1,109 @@
+package io.kestra.core.models.flows.check;
+
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Represents a check within a Kestra flow.
+ * <p>
+ * A {@code Check} defines a boolean condition that is evaluated when validating flow's inputs
+ * and before triggering an execution.
+ * <p>
+ * If the condition evaluates to {@code false}, the configured {@link Behavior}
+ * determines how the execution proceeds, and the {@link Style} determines how
+ * the message is visually presented in the UI.
+ * </p>
+ */
+@SuperBuilder
+@Getter
+@NoArgsConstructor
+public class Check {
+    
+    /**
+     * The condition to evaluate.
+     */
+    @NotNull
+    @NotEmpty
+    String condition;
+    
+    /**
+     * The message associated with this check, will be displayed when the condition evaluates to {@code false}.
+     */
+    @NotEmpty
+    String message;
+    
+    /**
+     * Defines the style of the message displayed in the UI when the condition evaluates to {@code false}.
+     */
+    Style style = Style.INFO;
+    
+    /**
+     * The behavior to apply when the condition evaluates to {@code false}.
+     */
+    Behavior behavior = Behavior.BLOCK_EXECUTION;
+    
+    /**
+     * The visual style used to display the message when the check fails.
+     */
+    public enum Style {
+        /**
+         * Display the message as an error.
+         */
+        ERROR,
+        /**
+         * Display the message as a success indicator.
+         */
+        SUCCESS,
+        /**
+         * Display the message as a warning.
+         */
+        WARNING,
+        /**
+         * Display the message as informational content.
+         */
+        INFO;
+    }
+    
+    /**
+     * Defines how the flow should behave when the condition evaluates to {@code false}.
+     */
+    public enum Behavior {
+        /**
+         * Block the creation of the execution.
+         */
+        BLOCK_EXECUTION,
+        /**
+         * Create the execution as failed.
+         */
+        FAIL_EXECUTION,
+        /**
+         * Create a new execution as a result of the check failing.
+         */
+        CREATE_EXECUTION;
+    }
+    
+    /**
+     * Resolves the effective behavior for a list of {@link Check}s based on priority.
+     *
+     * @param checks the list of checks whose behaviors are to be evaluated
+     * @return the highest-priority behavior, or {@code CREATE_EXECUTION} if the list is empty or only contains nulls
+     */
+    public static Check.Behavior resolveBehavior(List<Check> checks) {
+        if (checks == null || checks.isEmpty()) {
+            return Behavior.CREATE_EXECUTION;
+        }
+        
+        return checks.stream()
+            .map(Check::getBehavior)
+            .filter(Objects::nonNull).min(Comparator.comparingInt(Enum::ordinal))
+            .orElse(Behavior.CREATE_EXECUTION);
+    }
+    
+}

core/src/main/java/io/kestra/core/models/kv/PersistedKvMetadata.java

@@ -20,7 +20,6 @@ import java.util.Optional;
 @Slf4j
 @Getter
 @FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
-@AllArgsConstructor
 @ToString
 @EqualsAndHashCode
 public class PersistedKvMetadata implements DeletedInterface, TenantInterface, HasUID {
@@ -54,6 +53,19 @@ public class PersistedKvMetadata implements DeletedInterface, TenantInterface, H
 
     private boolean deleted;
 
+    public PersistedKvMetadata(String tenantId, String namespace, String name, String description, Integer version, boolean last, @Nullable Instant expirationDate, @Nullable Instant created, @Nullable Instant updated, boolean deleted) {
+        this.tenantId = tenantId;
+        this.namespace = namespace;
+        this.name = name;
+        this.description = description;
+        this.version = version;
+        this.last = last;
+        this.expirationDate = expirationDate;
+        this.created = Optional.ofNullable(created).orElse(Instant.now());
+        this.updated = updated;
+        this.deleted = deleted;
+    }
+
     public static PersistedKvMetadata from(String tenantId, KVEntry kvEntry) {
         return PersistedKvMetadata.builder()
             .tenantId(tenantId)
@@ -68,12 +80,15 @@ public class PersistedKvMetadata implements DeletedInterface, TenantInterface, H
     }
 
     public PersistedKvMetadata asLast() {
-        Instant saveDate = Instant.now();
-        return this.toBuilder().created(Optional.ofNullable(this.created).orElse(saveDate)).updated(saveDate).last(true).build();
+        return this.toBuilder().updated(Instant.now()).last(true).build();
+    }
+
+    public PersistedKvMetadata toDeleted() {
+        return this.toBuilder().updated(Instant.now()).deleted(true).build();
     }
 
     @Override
     public String uid() {
-        return IdUtils.fromParts(getTenantId(), getNamespace(), getName(), getVersion().toString());
+        return IdUtils.fromParts(getTenantId(), getNamespace(), getName(), String.valueOf(getVersion()));
     }
 }

core/src/main/java/io/kestra/core/models/namespaces/files/NamespaceFileMetadata.java

@@ -0,0 +1,132 @@
+package io.kestra.core.models.namespaces.files;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.TenantInterface;
+import io.kestra.core.storages.FileAttributes;
+import io.kestra.core.storages.NamespaceFile;
+import io.kestra.core.utils.IdUtils;
+import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.annotation.Nullable;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+import lombok.*;
+import lombok.experimental.FieldDefaults;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Instant;
+
+@Builder(toBuilder = true)
+@Slf4j
+@Getter
+@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
+@ToString
+@EqualsAndHashCode
+public class NamespaceFileMetadata implements DeletedInterface, TenantInterface, HasUID {
+    @With
+    @Hidden
+    @Pattern(regexp = "^[a-z0-9][a-z0-9_-]*")
+    private String tenantId;
+
+    @NotNull
+    private String namespace;
+
+    @NotNull
+    private String path;
+
+    private String parentPath;
+
+    @NotNull
+    private Integer version;
+
+    @Builder.Default
+    private boolean last = true;
+
+    @NotNull
+    private Long size;
+
+    @Builder.Default
+    private Instant created = Instant.now();
+
+    @Nullable
+    private Instant updated;
+
+    private boolean deleted;
+
+    @JsonCreator
+    public NamespaceFileMetadata(String tenantId, String namespace, String path, String parentPath, Integer version, boolean last, Long size, Instant created, @Nullable Instant updated, boolean deleted) {
+        this.tenantId = tenantId;
+        this.namespace = namespace;
+        this.path = path;
+        this.parentPath = parentPath(path);
+        this.version = version;
+        this.last = last;
+        this.size = size;
+        this.created = created;
+        this.updated = updated;
+        this.deleted = deleted;
+    }
+
+    public static String path(String path, boolean trailingSlash) {
+        if (trailingSlash && !path.endsWith("/")) {
+            return path + "/";
+        } else if (!trailingSlash && path.endsWith("/")) {
+            return path.substring(0, path.length() - 1);
+        }
+        return path;
+    }
+
+    public String path(boolean trailingSlash) {
+        return path(this.path, trailingSlash);
+    }
+
+    public static String parentPath(String path) {
+        String withoutTrailingSlash = path.endsWith("/") ? path.substring(0, path.length() - 1) : path;
+        // The parent path can't be set, it's always computed
+        return withoutTrailingSlash.contains("/") ?
+            withoutTrailingSlash.substring(0, withoutTrailingSlash.lastIndexOf("/") + 1) :
+            null;
+    }
+
+    public static NamespaceFileMetadata of(String tenantId, NamespaceFile namespaceFile) {
+        return NamespaceFileMetadata.builder()
+            .tenantId(tenantId)
+            .namespace(namespaceFile.namespace())
+            .path(namespaceFile.path(true).toString())
+            .version(namespaceFile.version())
+            .build();
+    }
+
+    public static NamespaceFileMetadata of(String tenantId, String namespace, String path, FileAttributes fileAttributes) {
+        return NamespaceFileMetadata.builder()
+            .tenantId(tenantId)
+            .namespace(namespace)
+            .path(path)
+            .created(Instant.ofEpochMilli(fileAttributes.getCreationTime()))
+            .updated(Instant.ofEpochMilli(fileAttributes.getLastModifiedTime()))
+            .size(fileAttributes.getSize())
+            .version(1)
+            .build();
+    }
+
+    public NamespaceFileMetadata asLast() {
+        Instant saveDate = Instant.now();
+        return this.toBuilder().updated(saveDate).last(true).build();
+    }
+
+    public NamespaceFileMetadata toDeleted() {
+        return this.toBuilder().deleted(true).updated(Instant.now()).build();
+    }
+
+    @Override
+    public String uid() {
+        return IdUtils.fromParts(getTenantId(), getNamespace(), getPath(), String.valueOf(getVersion()));
+    }
+
+    @JsonIgnore
+    public boolean isDirectory() {
+        return this.path.endsWith("/");
+    }
+}

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

@@ -35,7 +35,6 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @JsonDeserialize(using = Property.PropertyDeserializer.class)
 @JsonSerialize(using = Property.PropertySerializer.class)
 @Builder
-@NoArgsConstructor
 @AllArgsConstructor(access = AccessLevel.PACKAGE)
 @Schema(
     oneOf = {
@@ -51,6 +50,7 @@ public class Property<T> {
         .copy()
         .configure(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS, false);
 
+    private final boolean skipCache;
     private String expression;
     private T value;
 
@@ -60,13 +60,23 @@ public class Property<T> {
     @Deprecated
     // Note: when not used, this constructor would not be deleted but made private so it can only be used by ofExpression(String) and the deserializer
     public Property(String expression) {
-        this.expression = expression;
+        this(expression, false);
     }
 
+    private Property(String expression, boolean skipCache) {
+        this.expression = expression;
+        this.skipCache = skipCache;
+    }
+
+    /**
+     * @deprecated use {@link #ofValue(Object)} instead.
+     */
     @VisibleForTesting
+    @Deprecated
     public Property(Map<?, ?> map) {
         try {
             expression = MAPPER.writeValueAsString(map);
+            this.skipCache = false;
         } catch (JsonProcessingException e) {
             throw new IllegalArgumentException(e);
         }
@@ -79,14 +89,11 @@ public class Property<T> {
     /**
      * Returns a new {@link Property} with no cached rendered value,
      * so that the next render will evaluate its original Pebble expression.
-     * <p>
-     * The returned property will still cache its rendered result.
-     * To re-evaluate on a subsequent render, call {@code skipCache()} again.
      *
      * @return a new {@link Property} without a pre-rendered value
      */
     public Property<T> skipCache() {
-        return Property.ofExpression(expression);
+        return new Property<>(expression, true);
     }
 
     /**
@@ -133,6 +140,7 @@ public class Property<T> {
 
     /**
      * Build a new Property object with a Pebble expression.<br>
+     * This property object will not cache its rendered value.
      * <p>
      * Use {@link #ofValue(Object)} to build a property with a value instead.
      */
@@ -142,11 +150,11 @@ public class Property<T> {
             throw new IllegalArgumentException("'expression' must be a valid Pebble expression");
         }
 
-        return new Property<>(expression);
+        return new Property<>(expression, true);
     }
 
     /**
-     * Render a property then convert it to its target type.<br>
+     * Render a property, then convert it to its target type.<br>
      * <p>
      * This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
      *
@@ -164,7 +172,7 @@ public class Property<T> {
      * @see io.kestra.core.runners.RunContextProperty#as(Class, Map)
      */
     public static <T> T as(Property<T> property, PropertyContext context, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
-        if (property.value == null) {
+        if (property.skipCache || property.value == null) {
             String rendered = context.render(property.expression, variables);
             property.value = MAPPER.convertValue(rendered, clazz);
         }
@@ -192,7 +200,7 @@ public class Property<T> {
      */
     @SuppressWarnings("unchecked")
     public static <T, I> T asList(Property<T> property, PropertyContext context, Class<I> itemClazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
-        if (property.value == null) {
+        if (property.skipCache || property.value == null) {
             JavaType type = MAPPER.getTypeFactory().constructCollectionLikeType(List.class, itemClazz);
             try {
                 String trimmedExpression = property.expression.trim();
@@ -244,7 +252,7 @@ public class Property<T> {
      */
     @SuppressWarnings({"rawtypes", "unchecked"})
     public static <T, K, V> T asMap(Property<T> property, RunContext runContext, Class<K> keyClass, Class<V> valueClass, Map<String, Object> variables) throws IllegalVariableEvaluationException {
-        if (property.value == null) {
+        if (property.skipCache || property.value == null) {
             JavaType targetMapType = MAPPER.getTypeFactory().constructMapType(Map.class, keyClass, valueClass);
 
             try {

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

@@ -4,10 +4,8 @@ import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.tasks.runners.TaskLogLineMatcher.TaskLogMatch;
-import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
-import io.kestra.core.services.FlowService;
 import jakarta.validation.constraints.NotNull;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
@@ -38,6 +36,7 @@ import static io.kestra.core.utils.Rethrow.throwConsumer;
 abstract public class PluginUtilsService {
 
     private static final TypeReference<Map<String, String>> MAP_TYPE_REFERENCE = new TypeReference<>() {};
+    private static final TaskLogLineMatcher LOG_LINE_MATCHER = new TaskLogLineMatcher();
 
     public static Map<String, String> createOutputFiles(
         Path tempDirectory,
@@ -170,12 +169,9 @@ abstract public class PluginUtilsService {
     }
 
     public static Map<String, Object> parseOut(String line, Logger logger, RunContext runContext, boolean isStdErr, Instant customInstant) {
-
-        TaskLogLineMatcher logLineMatcher = ((DefaultRunContext) runContext).getApplicationContext().getBean(TaskLogLineMatcher.class);
-
         Map<String, Object> outputs = new HashMap<>();
         try {
-            Optional<TaskLogMatch> matches = logLineMatcher.matches(line, logger, runContext, customInstant);
+            Optional<TaskLogMatch> matches = LOG_LINE_MATCHER.matches(line, logger, runContext, customInstant);
             if (matches.isPresent()) {
                 TaskLogMatch taskLogMatch = matches.get();
                 outputs.putAll(taskLogMatch.outputs());
@@ -215,8 +211,7 @@ abstract public class PluginUtilsService {
                 realNamespace = runContext.render(namespace);
                 realFlowId = runContext.render(flowId);
                 // validate that the flow exists: a.k.a access is authorized by this namespace
-                FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
-                flowService.checkAllowedNamespace(flowInfo.tenantId(), realNamespace, flowInfo.tenantId(), flowInfo.namespace());
+                runContext.acl().allowNamespace(realNamespace).check();
             } else if (namespace != null || flowId != null) {
                 throw new IllegalArgumentException("Both `namespace` and `flowId` must be set when `executionId` is set.");
             } else {

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

@@ -27,7 +27,6 @@ import static io.kestra.core.runners.RunContextLogger.ORIGINAL_TIMESTAMP_KEY;
  * ::{"outputs":{"key":"value"}}::
  * }</pre>
  */
-@Singleton
 public class TaskLogLineMatcher {
 
     protected static final Pattern LOG_DATA_SYNTAX = Pattern.compile("^::(\\{.*})::$");
@@ -108,4 +107,4 @@ public class TaskLogLineMatcher {
         String message
     ) {
     }
-}
+}

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

@@ -82,6 +82,12 @@ abstract public class AbstractTrigger implements TriggerInterface {
     @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
     private boolean failOnTriggerError = false;
 
+    @PluginProperty(group = PluginProperty.CORE_GROUP)
+    @Schema(
+        title = "Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress."
+    )
+    private boolean allowConcurrent = false;
+
     /**
      * For backward compatibility: we rename minLogLevel to logLevel.
      * @deprecated use {@link #logLevel} instead

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

@@ -1,22 +1,37 @@
 package io.kestra.core.models.triggers;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.runners.RunContext;
+import io.swagger.v3.oas.annotations.media.Schema;
 
 import java.time.ZonedDateTime;
+import java.util.Map;
 
 public interface Schedulable extends PollingTriggerInterface{
     String PLUGIN_PROPERTY_RECOVER_MISSED_SCHEDULES = "recoverMissedSchedules";
 
+    @Schema(
+        title = "The inputs to pass to the scheduled flow"
+    )
+    @PluginProperty(dynamic = true)
+    Map<String, Object> getInputs();
+
+    @Schema(
+        title = "Action to take in the case of missed schedules",
+        description = "`ALL` will recover all missed schedules, `LAST`  will only recovered the last missing one, `NONE` will not recover any missing schedule.\n" +
+            "The default is `ALL` unless a different value is configured using the global plugin configuration."
+    )
+    @PluginProperty
+    RecoverMissedSchedules getRecoverMissedSchedules();
+    
     /**
      * Compute the previous evaluation of a trigger.
      * This is used when a trigger misses some schedule to compute the next date to evaluate in the past.
      */
     ZonedDateTime previousEvaluationDate(ConditionContext conditionContext) throws IllegalVariableEvaluationException;
-
-    RecoverMissedSchedules getRecoverMissedSchedules();
-
+    
     /**
      * Load the default RecoverMissedSchedules from plugin property, or else ALL.
      */

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

@@ -172,7 +172,7 @@ public class Trigger extends TriggerContext implements HasUID {
 
         if (abstractTrigger instanceof PollingTriggerInterface pollingTriggerInterface) {
             try {
-                nextDate = pollingTriggerInterface.nextEvaluationDate(conditionContext, Optional.empty());
+                nextDate = pollingTriggerInterface.nextEvaluationDate(conditionContext, lastTrigger);
             } catch (InvalidTriggerConfigurationException e) {
                 disabled = true;
             }

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

@@ -6,12 +6,9 @@ import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.ExecutionTrigger;
 import io.kestra.core.models.tasks.Output;
 import io.kestra.core.models.flows.State;
-import io.kestra.core.runners.DefaultRunContext;
-import io.kestra.core.runners.FlowInputOutput;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.ListUtils;
-import java.time.ZonedDateTime;
 import java.util.*;
 
 public abstract class TriggerService {
@@ -51,58 +48,6 @@ public abstract class TriggerService {
         return generateExecution(IdUtils.create(), trigger, context, executionTrigger, conditionContext);
     }
 
-    public static Execution generateScheduledExecution(
-        AbstractTrigger trigger,
-        ConditionContext conditionContext,
-        TriggerContext context,
-        List<Label> labels,
-        Map<String, Object> inputs,
-        Map<String, Object> variables,
-        Optional<ZonedDateTime> scheduleDate
-    ) {
-        RunContext runContext = conditionContext.getRunContext();
-        ExecutionTrigger executionTrigger = ExecutionTrigger.of(trigger, variables);
-
-        List<Label> executionLabels = new ArrayList<>(ListUtils.emptyOnNull(labels));
-        if (executionLabels.stream().noneMatch(label -> Label.CORRELATION_ID.equals(label.key()))) {
-            // add a correlation ID if none exist
-            executionLabels.add(new Label(Label.CORRELATION_ID, runContext.getTriggerExecutionId()));
-        }
-        Execution execution = Execution.builder()
-            .id(runContext.getTriggerExecutionId())
-            .tenantId(context.getTenantId())
-            .namespace(context.getNamespace())
-            .flowId(context.getFlowId())
-            .flowRevision(conditionContext.getFlow().getRevision())
-            .variables(conditionContext.getFlow().getVariables())
-            .labels(executionLabels)
-            .state(new State())
-            .trigger(executionTrigger)
-            .scheduleDate(scheduleDate.map(date -> date.toInstant()).orElse(null))
-            .build();
-
-        Map<String, Object> allInputs = new HashMap<>();
-        // add flow inputs with default value
-        var flow = conditionContext.getFlow();
-        if (flow.getInputs() != null) {
-            flow.getInputs().stream()
-                .filter(input -> input.getDefaults() != null)
-                .forEach(input -> allInputs.put(input.getId(), input.getDefaults()));
-        }
-
-        if (inputs != null) {
-            allInputs.putAll(inputs);
-        }
-
-        // add inputs and inject defaults
-        if (!allInputs.isEmpty()) {
-            FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class);
-            execution = execution.withInputs(flowInputOutput.readExecutionInputs(conditionContext.getFlow(), execution, allInputs));
-        }
-
-        return execution;
-    }
-
     private static Execution generateExecution(
         String id,
         AbstractTrigger trigger,
@@ -111,6 +56,7 @@ public abstract class TriggerService {
         ConditionContext conditionContext
     ) {
         List<Label> executionLabels = new ArrayList<>(ListUtils.emptyOnNull(trigger.getLabels()));
+        executionLabels.add(new Label(Label.FROM, "trigger"));
         if (executionLabels.stream().noneMatch(label -> Label.CORRELATION_ID.equals(label.key()))) {
             // add a correlation ID if none exist
             executionLabels.add(new Label(Label.CORRELATION_ID, id));

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

@@ -2,7 +2,6 @@ package io.kestra.core.repositories;
 
 import io.kestra.core.models.QueryFilter;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.executions.statistics.DailyExecutionStatistics;
 import io.kestra.core.models.executions.statistics.ExecutionCount;
 import io.kestra.core.models.executions.statistics.Flow;
@@ -94,6 +93,8 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
 
     Flux<Execution> findAllAsync(@Nullable String tenantId);
 
+    Flux<Execution> findAsync(String tenantId, List<QueryFilter> filters);
+
     Execution delete(Execution execution);
 
     Integer purge(Execution execution);

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

@@ -8,6 +8,7 @@ import io.kestra.plugin.core.dashboard.data.Flows;
 import io.micronaut.data.model.Pageable;
 import jakarta.annotation.Nullable;
 import jakarta.validation.ConstraintViolationException;
+import reactor.core.publisher.Flux;
 
 import java.util.List;
 import java.util.Optional;
@@ -158,6 +159,8 @@ public interface FlowRepositoryInterface extends QueryBuilderInterface<Flows.Fie
             .toList();
     }
 
+    Flux<Flow> findAsync(String tenantId, List<QueryFilter> filters);
+
     FlowWithSource create(GenericFlow flow);
 
     FlowWithSource update(GenericFlow flow, FlowInterface previous) throws ConstraintViolationException;

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

@@ -0,0 +1,46 @@
+package io.kestra.core.repositories;
+
+import io.kestra.core.models.FetchVersion;
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
+import io.micronaut.data.model.Pageable;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+
+public interface NamespaceFileMetadataRepositoryInterface extends SaveRepositoryInterface<NamespaceFileMetadata> {
+    Optional<NamespaceFileMetadata> findByPath(
+        String tenantId,
+        String namespace,
+        String path
+    ) throws IOException;
+
+    default ArrayListTotal<NamespaceFileMetadata> find(
+        Pageable pageable,
+        String tenantId,
+        List<QueryFilter> filters,
+        boolean allowDeleted
+    ) {
+        return this.find(pageable, tenantId, filters, allowDeleted, FetchVersion.LATEST);
+    }
+
+    ArrayListTotal<NamespaceFileMetadata> find(
+        Pageable pageable,
+        String tenantId,
+        List<QueryFilter> filters,
+        boolean allowDeleted,
+        FetchVersion fetchBehavior
+    );
+
+    default NamespaceFileMetadata delete(NamespaceFileMetadata namespaceFileMetadata) throws IOException {
+        return this.save(namespaceFileMetadata.toBuilder().deleted(true).build());
+    }
+
+    /**
+     * Purge (hard delete) a list of namespace files metadata. If no version is specified, all versions are purged.
+     * @param namespaceFilesMetadata the list of namespace files metadata to purge
+     * @return the number of purged namespace files metadata
+     */
+    Integer purge(List<NamespaceFileMetadata> namespaceFilesMetadata);
+}

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

@@ -1,10 +1,10 @@
 package io.kestra.core.repositories;
 
 import io.kestra.core.models.Setting;
+import jakarta.validation.ConstraintViolationException;
 
 import java.util.List;
 import java.util.Optional;
-import jakarta.validation.ConstraintViolationException;
 
 public interface SettingRepositoryInterface {
     Optional<Setting> findByKey(String key);
@@ -13,5 +13,7 @@ public interface SettingRepositoryInterface {
 
     Setting save(Setting setting) throws ConstraintViolationException;
 
+    Setting internalSave(Setting setting) throws ConstraintViolationException;
+
     Setting delete(Setting setting);
 }

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

@@ -16,8 +16,8 @@ import java.util.function.Function;
 public interface TriggerRepositoryInterface extends QueryBuilderInterface<Triggers.Fields> {
     Optional<Trigger> findLast(TriggerContext trigger);
 
-    Optional<Trigger> findByExecution(Execution execution);
-
+    Optional<Trigger> findByUid(String uid);
+    
     List<Trigger> findAll(String tenantId);
 
     List<Trigger> findAllForAllTenants();
@@ -43,9 +43,9 @@ public interface TriggerRepositoryInterface extends QueryBuilderInterface<Trigge
 
     /**
      * Find all triggers that match the query, return a flux of triggers
-     * as the search is not paginated
      */
-    Flux<Trigger> find(String tenantId, List<QueryFilter> filters);
+    Flux<Trigger> findAsync(String tenantId, List<QueryFilter> filters);
+
 
     default Function<String, String> sortMapping() throws IllegalArgumentException {
         return Function.identity();

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

@@ -0,0 +1,50 @@
+package io.kestra.core.runners;
+
+import javax.annotation.CheckReturnValue;
+import java.util.List;
+
+/**
+ * Check if the current taskrun has access to the requested resources.
+ *
+ * <p>
+ * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
+ *
+ * @see AllowedResources
+ */
+public interface AclChecker {
+
+    /**Tasks that need to access resources outside their namespace should use this interface to check ACL (Allowed namespaces in EE).
+     * Allow all namespaces.
+     * <p>
+     * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
+     */
+    @CheckReturnValue
+    AllowedResources allowAllNamespaces();
+
+    /**
+     * Allow only the given namespace.
+     * <p>
+     * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
+     */
+    @CheckReturnValue
+    AllowedResources allowNamespace(String namespace);
+
+    /**
+     * Allow only the given namespaces.
+     * <p>
+     * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
+     */
+    @CheckReturnValue
+    AllowedResources allowNamespaces(List<String> namespaces);
+
+    /**
+     * Represents a set of allowed resources.
+     * Tasks that need to access resources outside their namespace should call the <code>check()</code> method to check the ACL (Allowed namespaces in EE).
+     */
+    interface AllowedResources {
+        /**
+         * Check if the current taskrun has access to the requested resources.
+         */
+        void check();
+    }
+}

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

@@ -0,0 +1,86 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.services.NamespaceService;
+import io.micronaut.context.ApplicationContext;
+
+import java.util.List;
+import java.util.Objects;
+
+class AclCheckerImpl implements AclChecker {
+    private final NamespaceService namespaceService;
+    private final RunContext.FlowInfo flowInfo;
+
+    AclCheckerImpl(ApplicationContext applicationContext, RunContext.FlowInfo flowInfo) {
+        this.namespaceService = applicationContext.getBean(NamespaceService.class);
+        this.flowInfo = flowInfo;
+    }
+
+    @Override
+    public AllowedResources allowAllNamespaces() {
+        return new AllowAllNamespaces(flowInfo, namespaceService);
+    }
+
+    @Override
+    public AllowedResources allowNamespace(String namespace) {
+        return new AllowNamespace(flowInfo, namespaceService, namespace);
+    }
+
+    @Override
+    public AllowedResources allowNamespaces(List<String> namespaces) {
+        return new AllowNamespaces(flowInfo, namespaceService, namespaces);
+    }
+
+
+    static class AllowAllNamespaces implements AllowedResources {
+        private final RunContext.FlowInfo flowInfo;
+        private final NamespaceService namespaceService;
+
+        AllowAllNamespaces(RunContext.FlowInfo flowInfo, NamespaceService namespaceService) {
+            this.flowInfo = Objects.requireNonNull(flowInfo);
+            this.namespaceService = Objects.requireNonNull(namespaceService);
+        }
+
+        @Override
+        public void check() {
+            this.namespaceService.checkAllowedAllNamespaces(flowInfo.tenantId(), flowInfo.tenantId(), flowInfo.namespace());
+        }
+    }
+
+    static class AllowNamespace implements AllowedResources {
+        private final RunContext.FlowInfo flowInfo;
+        private final NamespaceService namespaceService;
+        private final String namespace;
+
+        public AllowNamespace(RunContext.FlowInfo flowInfo, NamespaceService namespaceService, String namespace) {
+            this.flowInfo = Objects.requireNonNull(flowInfo);
+            this.namespaceService = Objects.requireNonNull(namespaceService);
+            this.namespace = Objects.requireNonNull(namespace);
+        }
+
+        @Override
+        public void check() {
+            namespaceService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace());
+        }
+    }
+
+    static class AllowNamespaces implements AllowedResources {
+        private final RunContext.FlowInfo flowInfo;
+        private final NamespaceService namespaceService;
+        private final List<String> namespaces;
+
+        AllowNamespaces(RunContext.FlowInfo flowInfo, NamespaceService namespaceService, List<String> namespaces) {
+            this.flowInfo = Objects.requireNonNull(flowInfo);
+            this.namespaceService = Objects.requireNonNull(namespaceService);
+            this.namespaces = Objects.requireNonNull(namespaces);
+
+            if (namespaces.isEmpty()) {
+                throw new IllegalArgumentException("At least one namespace must be provided");
+            }
+        }
+
+        @Override
+        public void check() {
+            namespaces.forEach(namespace -> namespaceService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace()));
+        }
+    }
+}

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

@@ -6,10 +6,12 @@ import com.google.common.base.CaseFormat;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.Plugin;
 import io.kestra.core.models.executions.AbstractMetricEntry;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.plugins.PluginConfigurations;
 import io.kestra.core.services.KVStoreService;
 import io.kestra.core.storages.Storage;
 import io.kestra.core.storages.StorageInterface;
@@ -123,7 +125,12 @@ public class DefaultRunContext extends RunContext {
         this.traceParent = traceParent;
     }
 
+    /**
+     * @deprecated Plugin should not use the ApplicationContext anymore, and neither should they cast to this implementation.
+     *             Plugin should instead rely on supported API only.
+     */
     @JsonIgnore
+    @Deprecated(since = "1.2.0", forRemoval = true)
     public ApplicationContext getApplicationContext() {
         return applicationContext;
     }
@@ -230,6 +237,14 @@ public class DefaultRunContext extends RunContext {
         return runContext;
     }
 
+    @Override
+    public RunContext cloneForPlugin(Plugin plugin) {
+        PluginConfigurations pluginConfigurations = applicationContext.getBean(PluginConfigurations.class);
+        DefaultRunContext runContext = clone();
+        runContext.pluginConfiguration = pluginConfigurations.getConfigurationByPluginTypeOrAliases(plugin.getType(), plugin.getClass());
+        return runContext;
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -574,11 +589,21 @@ public class DefaultRunContext extends RunContext {
         return isInitialized.get();
     }
 
+    @Override
+    public AclChecker acl() {
+        return new AclCheckerImpl(this.applicationContext, flowInfo());
+    }
+
     @Override
     public LocalPath localPath() {
         return localPath;
     }
 
+    @Override
+    public InputAndOutput inputAndOutput() {
+        return new InputAndOutputImpl(this.applicationContext, this);
+    }
+
     /**
      * Builder class for constructing new {@link DefaultRunContext} objects.
      */

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

@@ -53,12 +53,10 @@ public final class ExecutableUtils {
     }
 
     public static SubflowExecutionResult subflowExecutionResult(TaskRun parentTaskrun, Execution execution) {
-        List<TaskRunAttempt> attempts = parentTaskrun.getAttempts() == null ? new ArrayList<>() : new ArrayList<>(parentTaskrun.getAttempts());
-        attempts.add(TaskRunAttempt.builder().state(parentTaskrun.getState()).build());
         return SubflowExecutionResult.builder()
             .executionId(execution.getId())
             .state(parentTaskrun.getState().getCurrent())
-            .parentTaskRun(parentTaskrun.withAttempts(attempts))
+            .parentTaskRun(parentTaskrun.addAttempt(TaskRunAttempt.builder().state(parentTaskrun.getState()).build()))
             .build();
     }
 
@@ -191,12 +189,11 @@ public final class ExecutableUtils {
                 variables.put("taskRunIteration", currentTaskRun.getIteration());
             }
 
-            FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class);
             Instant scheduleOnDate = runContext.render(scheduleDate).as(ZonedDateTime.class).map(date -> date.toInstant()).orElse(null);
             Execution execution = Execution
                 .newExecution(
                     flow,
-                    (f, e) -> flowInputOutput.readExecutionInputs(f, e, inputs),
+                    (f, e) -> runContext.inputAndOutput().readInputs(f, e, inputs),
                     newLabels,
                     Optional.empty())
                 .withTrigger(ExecutionTrigger.builder()

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

@@ -3,13 +3,11 @@ package io.kestra.core.runners;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.kestra.core.encryption.EncryptionService;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
-import io.kestra.core.exceptions.KestraRuntimeException;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Data;
 import io.kestra.core.models.flows.DependsOn;
 import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.Input;
-import io.kestra.core.models.flows.Output;
 import io.kestra.core.models.flows.RenderableInput;
 import io.kestra.core.models.flows.Type;
 import io.kestra.core.models.flows.input.FileInput;
@@ -158,11 +156,7 @@ public class FlowInputOutput {
                             File tempFile = File.createTempFile(prefix, fileExtension);
                             try (var inputStream = fileUpload.getInputStream();
                                  var outputStream = new FileOutputStream(tempFile)) {
-                                long transferredBytes = inputStream.transferTo(outputStream);
-                                if (transferredBytes == 0) {
-                                    sink.error(new KestraRuntimeException("Can't upload file: " + fileUpload.getFilename()));
-                                    return;
-                                }
+                                inputStream.transferTo(outputStream);
                                 URI from = storageInterface.from(execution, inputId, fileName, tempFile);
                                 sink.next(Map.entry(inputId, from.toString()));
                             } finally {
@@ -382,11 +376,11 @@ public class FlowInputOutput {
 
     @SuppressWarnings("unchecked")
     private static <T> Object resolveDefaultPropertyAs(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
-        return Property.as((Property<T>) input.getDefaults(), renderer, clazz);
+        return Property.as((Property<T>) input.getDefaults().skipCache(), renderer, clazz);
     }
     @SuppressWarnings("unchecked")
     private static <T> Object resolveDefaultPropertyAsList(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
-        return Property.asList((Property<List<T>>) input.getDefaults(), renderer, clazz);
+        return Property.asList((Property<List<T>>) input.getDefaults().skipCache(), renderer, clazz);
     }
 
     private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies, final boolean decryptSecrets) {
@@ -502,8 +496,8 @@ public class FlowInputOutput {
                         yield storageInterface.from(execution, id, current.toString().substring(current.toString().lastIndexOf("/") + 1), new File(current.toString()));
                     }
                 }
-                case JSON -> JacksonMapper.toObject(current.toString());
-                case YAML -> YAML_MAPPER.readValue(current.toString(), JacksonMapper.OBJECT_TYPE_REFERENCE);
+                case JSON -> (current instanceof Map || current instanceof Collection<?>) ? current :  JacksonMapper.toObject(current.toString());
+                case YAML -> (current instanceof Map || current instanceof Collection<?>) ? current : YAML_MAPPER.readValue(current.toString(), JacksonMapper.OBJECT_TYPE_REFERENCE);
                 case URI -> {
                     Matcher matcher = URI_PATTERN.matcher(current.toString());
                     if (matcher.matches()) {
@@ -543,30 +537,6 @@ public class FlowInputOutput {
         }
     }
 
-    public static Map<String, Object> renderFlowOutputs(List<Output> outputs, RunContext runContext) throws IllegalVariableEvaluationException {
-        if (outputs == null) return Map.of();
-
-        // render required outputs
-        Map<String, Object> outputsById = outputs
-            .stream()
-            .filter(output -> output.getRequired() == null || output.getRequired())
-            .collect(HashMap::new, (map, entry) -> map.put(entry.getId(), entry.getValue()), Map::putAll);
-        outputsById = runContext.render(outputsById);
-
-        // render optional outputs one by one to catch, log, and skip any error.
-        for (io.kestra.core.models.flows.Output output : outputs) {
-            if (Boolean.FALSE.equals(output.getRequired())) {
-                try {
-                    outputsById.putAll(runContext.render(Map.of(output.getId(), output.getValue())));
-                } catch (Exception e) {
-                    runContext.logger().warn("Failed to render optional flow output '{}'. Output is ignored.", output.getId(), e);
-                    outputsById.put(output.getId(), null);
-                }
-            }
-        }
-        return outputsById;
-    }
-
     /**
      * Mutable wrapper to hold a flow's input, and it's resolved value.
      */

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

@@ -11,6 +11,7 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.utils.ListUtils;
 import io.kestra.plugin.core.flow.Dag;
 
 import java.util.*;
@@ -143,6 +144,13 @@ public class FlowableUtils {
             return Collections.emptyList();
         }
 
+        // have submitted, leave
+        Optional<TaskRun> lastSubmitted = execution.findLastSubmitted(taskRuns);
+        if (lastSubmitted.isPresent()) {
+            return Collections.emptyList();
+        }
+
+
         // last success, find next
         Optional<TaskRun> lastTerminated = execution.findLastTerminated(taskRuns);
         if (lastTerminated.isPresent()) {
@@ -150,14 +158,41 @@ public class FlowableUtils {
 
             if (currentTasks.size() > lastIndex + 1) {
                 return Collections.singletonList(currentTasks.get(lastIndex + 1).toNextTaskRunIncrementIteration(execution, parentTaskRun.getIteration()));
-            } else {
-                return Collections.singletonList(currentTasks.getFirst().toNextTaskRunIncrementIteration(execution, parentTaskRun.getIteration()));
             }
         }
 
         return Collections.emptyList();
     }
 
+    public static Optional<State.Type> resolveSequentialState(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        List<ResolvedTask> _finally,
+        TaskRun parentTaskRun,
+        RunContext runContext,
+        boolean allowFailure,
+        boolean allowWarning
+    ) {
+        if (ListUtils.emptyOnNull(tasks).stream()
+            .filter(resolvedTask -> !resolvedTask.getTask().getDisabled())
+            .findAny()
+            .isEmpty()) {
+            return Optional.of(State.Type.SUCCESS);
+        }
+
+        return resolveState(
+            execution,
+            tasks,
+            errors,
+            _finally,
+            parentTaskRun,
+            runContext,
+            allowFailure,
+            allowWarning
+        );
+    }
+
     public static Optional<State.Type> resolveState(
         Execution execution,
         List<ResolvedTask> tasks,
@@ -213,7 +248,7 @@ public class FlowableUtils {
             }
         } else {
             // first call, the error flow is not ready, we need to notify the parent task that can be failed to init error flows
-            if (execution.hasFailed(tasks, parentTaskRun) || terminalState == State.Type.FAILED) {
+            if (execution.hasFailedNoRetry(tasks, parentTaskRun) || terminalState == State.Type.FAILED) {
                 return Optional.of(execution.guessFinalState(tasks, parentTaskRun, allowFailure, allowWarning, terminalState));
             }
         }

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

@@ -0,0 +1,29 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.FlowInterface;
+import io.kestra.core.models.flows.Output;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * InputAndOutput could be used to work with flow execution inputs and outputs.
+ */
+public interface InputAndOutput {
+    /**
+     * Reads the inputs of a flow execution.
+     */
+    Map<String, Object> readInputs(FlowInterface flow, Execution execution, Map<String, Object> inputs);
+
+    /**
+     * Processes the outputs of a flow execution (parse them based on their types).
+     */
+    Map<String, Object> typedOutputs(FlowInterface flow, Execution execution, Map<String, Object> rOutputs);
+
+    /**
+     * Render flow execution outputs.
+     */
+    Map<String, Object> renderOutputs(List<Output> outputs) throws IllegalVariableEvaluationException;
+}

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

@@ -0,0 +1,56 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.FlowInterface;
+import io.kestra.core.models.flows.Output;
+import io.micronaut.context.ApplicationContext;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class InputAndOutputImpl implements InputAndOutput {
+    private final FlowInputOutput flowInputOutput;
+    private final RunContext runContext;
+
+    InputAndOutputImpl(ApplicationContext applicationContext, RunContext runContext) {
+        this.flowInputOutput = applicationContext.getBean(FlowInputOutput.class);
+        this.runContext = runContext;
+    }
+
+    @Override
+    public Map<String, Object> readInputs(FlowInterface flow, Execution execution, Map<String, Object> inputs) {
+        return flowInputOutput.readExecutionInputs(flow, execution, inputs);
+    }
+
+    @Override
+    public Map<String, Object> typedOutputs(FlowInterface flow, Execution execution, Map<String, Object> rOutputs) {
+        return flowInputOutput.typedOutputs(flow, execution, rOutputs);
+    }
+
+    @Override
+    public Map<String, Object> renderOutputs(List<Output> outputs) throws IllegalVariableEvaluationException {
+        if (outputs == null) return Map.of();
+
+        // render required outputs
+        Map<String, Object> outputsById = outputs
+            .stream()
+            .filter(output -> output.getRequired() == null || output.getRequired())
+            .collect(HashMap::new, (map, entry) -> map.put(entry.getId(), entry.getValue()), Map::putAll);
+        outputsById = runContext.render(outputsById);
+
+        // render optional outputs one by one to catch, log, and skip any error.
+        for (io.kestra.core.models.flows.Output output : outputs) {
+            if (Boolean.FALSE.equals(output.getRequired())) {
+                try {
+                    outputsById.putAll(runContext.render(Map.of(output.getId(), output.getValue())));
+                } catch (Exception e) {
+                    runContext.logger().warn("Failed to render optional flow output '{}'. Output is ignored.", output.getId(), e);
+                    outputsById.put(output.getId(), null);
+                }
+            }
+        }
+        return outputsById;
+    }
+}

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

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.encryption.EncryptionService;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.Plugin;
 import io.kestra.core.models.executions.AbstractMetricEntry;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.property.PropertyContext;
@@ -192,5 +193,27 @@ public abstract class RunContext implements PropertyContext {
     public record FlowInfo(String tenantId, String namespace, String id, Integer revision) {
     }
 
+    /**
+     * @deprecated there is no legitimate use case of this method outside the run context internal self-usage, so it should not be part of the interface
+     */
+    @Deprecated(since = "1.2.0", forRemoval = true)
     public abstract boolean isInitialized();
+
+    /**
+     * Get access to the ACL checker.
+     * Plugins are responsible for using the ACL checker when they access restricted resources, for example,
+     * when Namespace ACLs are used (EE).
+     */
+    public abstract AclChecker acl();
+
+    /**
+     * Clone this run context for a specific plugin.
+     * @return a new run context with the plugin configuration of the given plugin.
+     */
+    public abstract RunContext cloneForPlugin(Plugin plugin);
+
+    /**
+     * @return an InputAndOutput that can be used to work with inputs and outputs.
+     */
+    public abstract InputAndOutput inputAndOutput();
 }

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

@@ -12,9 +12,10 @@ import io.kestra.core.models.property.PropertyContext;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.plugins.PluginConfigurations;
-import io.kestra.core.services.FlowService;
 import io.kestra.core.services.KVStoreService;
+import io.kestra.core.services.NamespaceService;
 import io.kestra.core.storages.InternalStorage;
+import io.kestra.core.storages.NamespaceFactory;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
 import io.micronaut.context.ApplicationContext;
@@ -48,7 +49,7 @@ public class RunContextFactory {
     protected StorageInterface storageInterface;
 
     @Inject
-    protected FlowService flowService;
+    protected NamespaceService namespaceService;
 
     @Inject
     protected MetricRegistry metricRegistry;
@@ -76,6 +77,9 @@ public class RunContextFactory {
     @Inject
     private KVStoreService kvStoreService;
 
+    @Inject
+    private NamespaceFactory namespaceFactory;
+
     // hacky
     public RunContextInitializer initializer() {
         return applicationContext.getBean(RunContextInitializer.class);
@@ -103,7 +107,7 @@ public class RunContextFactory {
             .withLogger(runContextLogger)
             // Execution
             .withPluginConfiguration(Map.of())
-            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forExecution(execution), storageInterface, flowService))
+            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forExecution(execution), storageInterface, namespaceService, namespaceFactory))
             .withVariableRenderer(variableRenderer)
             .withVariables(runVariableModifier.apply(
                     newRunVariablesBuilder()
@@ -133,7 +137,7 @@ public class RunContextFactory {
             .withLogger(runContextLogger)
             // Task
             .withPluginConfiguration(pluginConfigurations.getConfigurationByPluginTypeOrAliases(task.getType(), task.getClass()))
-            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, flowService))
+            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, namespaceService, namespaceFactory))
             .withVariables(newRunVariablesBuilder()
                 .withFlow(flow)
                 .withTask(task)
@@ -167,14 +171,16 @@ public class RunContextFactory {
             .build();
     }
 
-
-    @VisibleForTesting
     public RunContext of(final FlowInterface flow, final Map<String, Object> variables) {
         RunContextLogger runContextLogger = new RunContextLogger();
         return newBuilder()
             .withLogger(runContextLogger)
-            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forFlow(flow), storageInterface, flowService))
-            .withVariables(variables)
+            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forFlow(flow), storageInterface, namespaceService, namespaceFactory))
+            .withVariables(newRunVariablesBuilder()
+                .withFlow(flow)
+                .withVariables(variables)
+                .build(runContextLogger, PropertyContext.create(this.variableRenderer))
+            )
             .withSecretInputs(secretInputsFromFlow(flow))
             .build();
     }
@@ -212,7 +218,8 @@ public class RunContextFactory {
                     }
                 },
                 storageInterface,
-                flowService
+                namespaceService,
+                namespaceFactory
             ))
             .withVariables(variables)
             .withTask(task)

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

@@ -1,15 +1,14 @@
 package io.kestra.core.runners;
 
 import com.google.common.collect.Lists;
-import io.kestra.core.models.Plugin;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.tasks.Task;
-import io.kestra.core.models.tasks.runners.TaskRunner;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.plugins.PluginConfigurations;
-import io.kestra.core.services.FlowService;
+import io.kestra.core.services.NamespaceService;
 import io.kestra.core.storages.InternalStorage;
+import io.kestra.core.storages.NamespaceFactory;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.IdUtils;
@@ -44,25 +43,14 @@ public class RunContextInitializer {
     protected StorageInterface storageInterface;
 
     @Inject
-    protected FlowService flowService;
+    protected NamespaceFactory namespaceFactory;
+
+    @Inject
+    protected NamespaceService namespaceService;
 
     @Value("${kestra.encryption.secret-key}")
     protected Optional<String> secretKey;
 
-    /**
-     * Initializes the given {@link RunContext} for the given {@link Plugin}.
-     *
-     * @param runContext The {@link RunContext} to initialize.
-     * @param plugin The {@link TaskRunner} used for initialization.
-     * @return The {@link RunContext} to initialize
-     */
-    public DefaultRunContext forPlugin(final DefaultRunContext runContext,
-                                       final Plugin plugin) {
-        runContext.init(applicationContext);
-        runContext.setPluginConfiguration(pluginConfigurations.getConfigurationByPluginTypeOrAliases(plugin.getType(), plugin.getClass()));
-        return runContext;
-    }
-
     /**
      * Initializes the given {@link RunContext} for the given {@link WorkerTask} for executor.
      *
@@ -135,7 +123,7 @@ public class RunContextInitializer {
 
         runContext.setVariables(enrichedVariables);
         runContext.setPluginConfiguration(pluginConfigurations.getConfigurationByPluginTypeOrAliases(task.getType(), task.getClass()));
-        runContext.setStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, flowService));
+        runContext.setStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, namespaceService, namespaceFactory));
         runContext.setLogger(runContextLogger);
         runContext.setTask(task);
 
@@ -230,7 +218,8 @@ public class RunContextInitializer {
             runContextLogger.logger(),
             context,
             storageInterface,
-            flowService
+            namespaceService,
+            namespaceFactory
         );
 
         runContext.setLogger(runContextLogger);

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

@@ -55,11 +55,11 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
 
     public RunContextLogger(QueueInterface<LogEntry> logQueue, LogEntry logEntry, org.slf4j.event.Level loglevel, boolean logToFile) {
         if (logEntry.getTaskId() != null) {
-            this.loggerName = "flow." + logEntry.getFlowId() + "." + logEntry.getTaskId();
+            this.loggerName = baseLoggerName(logEntry) + "." + logEntry.getTaskId();
         } else if (logEntry.getTriggerId() != null) {
-            this.loggerName = "flow." + logEntry.getFlowId() + "." + logEntry.getTriggerId();
+            this.loggerName = baseLoggerName(logEntry) + "." + logEntry.getTriggerId();
         } else {
-            this.loggerName = "flow." + logEntry.getFlowId();
+            this.loggerName = baseLoggerName(logEntry);
         }
 
         this.logQueue = logQueue;
@@ -68,6 +68,10 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
         this.logToFile = logToFile;
     }
 
+    private String baseLoggerName(LogEntry logEntry) {
+        return "flow." + logEntry.getTenantId() + "." + logEntry.getNamespace() + "." + logEntry.getFlowId();
+    }
+
     private static List<LogEntry> logEntry(ILoggingEvent event, String message, org.slf4j.event.Level level, LogEntry logEntry) {
         Iterable<String> split;
 

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

@@ -2,6 +2,7 @@ package io.kestra.core.runners.pebble;
 
 import io.kestra.core.runners.VariableRenderer;
 import io.kestra.core.runners.pebble.functions.RenderingFunctionInterface;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.core.annotation.Nullable;
 import io.pebbletemplates.pebble.PebbleEngine;
@@ -18,35 +19,37 @@ import java.util.stream.Collectors;
 
 @Singleton
 public class PebbleEngineFactory {
-    
+
     private final ApplicationContext applicationContext;
     private final VariableRenderer.VariableConfiguration variableConfiguration;
-    
+    private final MeterRegistry meterRegistry;
+
     @Inject
-    public PebbleEngineFactory(ApplicationContext applicationContext, @Nullable VariableRenderer.VariableConfiguration variableConfiguration) {
+    public PebbleEngineFactory(ApplicationContext applicationContext, @Nullable VariableRenderer.VariableConfiguration variableConfiguration, MeterRegistry meterRegistry) {
         this.applicationContext = applicationContext;
         this.variableConfiguration = variableConfiguration;
+        this.meterRegistry = meterRegistry;
     }
-    
+
     public PebbleEngine create() {
         PebbleEngine.Builder builder = newPebbleEngineBuilder();
         this.applicationContext.getBeansOfType(Extension.class).forEach(builder::extension);
         return builder.build();
     }
-    
+
     public PebbleEngine createWithMaskedFunctions(VariableRenderer renderer, final List<String> functionsToMask) {
-        
+
         PebbleEngine.Builder builder = newPebbleEngineBuilder();
-        
+
         this.applicationContext.getBeansOfType(Extension.class).stream()
             .map(e -> functionsToMask.stream().anyMatch(fun -> e.getFunctions().containsKey(fun))
                 ? extensionWithMaskedFunctions(renderer, e, functionsToMask)
                 : e)
             .forEach(builder::extension);
-        
+
         return builder.build();
     }
-    
+
     private PebbleEngine.Builder newPebbleEngineBuilder() {
         PebbleEngine.Builder builder = new PebbleEngine.Builder()
             .registerExtensionCustomizer(ExtensionCustomizer::new)
@@ -54,13 +57,15 @@ public class PebbleEngineFactory {
             .cacheActive(this.variableConfiguration.getCacheEnabled())
             .newLineTrimming(false)
             .autoEscaping(false);
-        
+
         if (this.variableConfiguration.getCacheEnabled()) {
-            builder = builder.templateCache(new PebbleLruCache(this.variableConfiguration.getCacheSize()));
+            PebbleLruCache cache = new PebbleLruCache(this.variableConfiguration.getCacheSize());
+            cache.register(meterRegistry);
+            builder = builder.templateCache(cache);
         }
         return builder;
     }
-    
+
     private Extension extensionWithMaskedFunctions(VariableRenderer renderer, Extension initialExtension, List<String> maskedFunctions) {
         return (Extension) Proxy.newProxyInstance(
             initialExtension.getClass().getClassLoader(),
@@ -74,16 +79,16 @@ public class PebbleEngineFactory {
                             } else if (RenderingFunctionInterface.class.isAssignableFrom(entry.getValue().getClass())) {
                                 return Map.entry(entry.getKey(), this.variableRendererProxy(renderer, entry.getValue()));
                             }
-                            
+
                             return entry;
                         }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
                 }
-                
+
                 return method.invoke(initialExtension, methodArgs);
             }
         );
     }
-    
+
     private Function variableRendererProxy(VariableRenderer renderer, Function initialFunction) {
         return (Function) Proxy.newProxyInstance(
             initialFunction.getClass().getClassLoader(),
@@ -96,7 +101,7 @@ public class PebbleEngineFactory {
             }
         );
     }
-    
+
     private Function maskedFunctionProxy(Function initialFunction) {
         return (Function) Proxy.newProxyInstance(
             initialFunction.getClass().getClassLoader(),

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

@@ -1,29 +1,29 @@
 package io.kestra.core.runners.pebble;
 
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.binder.cache.CaffeineCacheMetrics;
 import io.pebbletemplates.pebble.cache.PebbleCache;
 import io.pebbletemplates.pebble.template.PebbleTemplate;
-import lombok.extern.slf4j.Slf4j;
 
-import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 
-@Slf4j
 public class PebbleLruCache implements PebbleCache<Object, PebbleTemplate> {
-    Cache<Object, PebbleTemplate> cache;
+    private final Cache<Object, PebbleTemplate> cache;
 
     public PebbleLruCache(int maximumSize) {
-        cache = CacheBuilder.newBuilder()
+        cache = Caffeine.newBuilder()
             .initialCapacity(250)
             .maximumSize(maximumSize)
+            .recordStats()
             .build();
     }
 
     @Override
     public PebbleTemplate computeIfAbsent(Object key, Function<? super Object, ? extends PebbleTemplate> mappingFunction) {
         try {
-            return cache.get(key, () -> mappingFunction.apply(key));
+            return cache.get(key, mappingFunction);
         } catch (Exception e) {
             // we retry the mapping function in order to let the exception be thrown instead of being capture by cache
             return mappingFunction.apply(key);
@@ -34,4 +34,8 @@ public class PebbleLruCache implements PebbleCache<Object, PebbleTemplate> {
     public void invalidateAll() {
         cache.invalidateAll();
     }
+
+    public void register(MeterRegistry meterRegistry) {
+        CaffeineCacheMetrics.monitor(meterRegistry, cache, "pebble-template");
+    }
 }

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

@@ -2,11 +2,8 @@ package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.runners.LocalPath;
 import io.kestra.core.runners.LocalPathFactory;
-import io.kestra.core.services.FlowService;
-import io.kestra.core.storages.InternalNamespace;
-import io.kestra.core.storages.Namespace;
-import io.kestra.core.storages.StorageContext;
-import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.services.NamespaceService;
+import io.kestra.core.storages.*;
 import io.kestra.core.utils.Slugify;
 import io.micronaut.context.annotation.Value;
 import io.pebbletemplates.pebble.error.PebbleException;
@@ -36,7 +33,7 @@ abstract class AbstractFileFunction implements Function {
     private static final Pattern EXECUTION_FILE = Pattern.compile(".*/.*/executions/.*/tasks/.*/.*");
 
     @Inject
-    protected FlowService flowService;
+    protected NamespaceService namespaceService;
 
     @Inject
     protected StorageInterface storageInterface;
@@ -44,6 +41,9 @@ abstract class AbstractFileFunction implements Function {
     @Inject
     protected LocalPathFactory localPathFactory;
 
+    @Inject
+    protected NamespaceFactory namespaceFactory;
+
     @Value("${" + LocalPath.ENABLE_FILE_FUNCTIONS_CONFIG + ":true}")
     protected boolean enableFileProtocol;
 
@@ -81,23 +81,21 @@ abstract class AbstractFileFunction implements Function {
                 } else if (str.startsWith(LocalPath.FILE_PROTOCOL)) {
                     fileUri = URI.create(str);
                     namespace = checkEnabledLocalFileAndReturnNamespace(args, flow);
-                } else if(str.startsWith(Namespace.NAMESPACE_FILE_SCHEME)) {
-                    URI nsFileUri = URI.create(str);
-                    namespace = checkedAllowedNamespaceAndReturnNamespace(args, nsFileUri, tenantId, flow);
-                    InternalNamespace internalNamespace = new InternalNamespace(flow.get(TENANT_ID), namespace, storageInterface);
-                    fileUri = internalNamespace.get(Path.of(nsFileUri.getPath())).uri();
+                } else if (str.startsWith(Namespace.NAMESPACE_FILE_SCHEME)) {
+                    fileUri = URI.create(str);
+                    namespace = checkedAllowedNamespaceAndReturnNamespace(args, fileUri, tenantId, flow);
                 } else if (URI_PATTERN.matcher(str).matches()) {
                     // it is an unsupported URI
                     throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(str));
                 } else {
+                    fileUri = URI.create(Namespace.NAMESPACE_FILE_SCHEME + ":///" + str);
                     namespace = (String) Optional.ofNullable(args.get(NAMESPACE)).orElse(flow.get(NAMESPACE));
-                    fileUri = URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.namespaceFilePrefix(namespace) + "/" + str);
-                    flowService.checkAllowedNamespace(tenantId, namespace, tenantId, flow.get(NAMESPACE));
+                    namespaceService.checkAllowedNamespace(tenantId, namespace, tenantId, flow.get(NAMESPACE));
                 }
             } else {
                 throw new PebbleException(null, "Unable to read the file " + path, lineNumber, self.getName());
             }
-            return fileFunction(context, fileUri, namespace, tenantId);
+            return fileFunction(context, fileUri, namespace, tenantId, args);
         } catch (IOException e) {
             throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
         }
@@ -110,7 +108,7 @@ abstract class AbstractFileFunction implements Function {
 
     protected abstract String getErrorMessage();
 
-    protected abstract Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException;
+    protected abstract Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException;
 
     boolean isFileUriValid(String namespace, String flowId, String executionId, URI path) {
         // Internal storage URI should be: kestra:///$namespace/$flowId/executions/$executionId/tasks/$taskName/$taskRunId/$random.ion or kestra:///$namespace/$flowId/executions/$executionId/trigger/$triggerName/$random.ion
@@ -177,7 +175,7 @@ abstract class AbstractFileFunction implements Function {
         // 5. replace '/' with '.'
         namespace = namespace.replace("/", ".");
 
-        flowService.checkAllowedNamespace(tenantId, namespace, tenantId, fromNamespace);
+        namespaceService.checkAllowedNamespace(tenantId, namespace, tenantId, fromNamespace);
 
         return namespace;
     }
@@ -198,7 +196,7 @@ abstract class AbstractFileFunction implements Function {
         // we will transform nsfile URI into a kestra URI so it is handled seamlessly by all functions
         String customNs = Optional.ofNullable((String) args.get(NAMESPACE)).orElse(nsFileUri.getAuthority());
         if (customNs != null) {
-            flowService.checkAllowedNamespace(tenantId, customNs, tenantId, flow.get(NAMESPACE));
+            namespaceService.checkAllowedNamespace(tenantId, customNs, tenantId, flow.get(NAMESPACE));
         }
         return Optional.ofNullable(customNs).orElse(flow.get(NAMESPACE));
     }

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

@@ -3,7 +3,7 @@ package io.kestra.core.runners.pebble.functions;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.tasks.retrys.Exponential;
 import io.kestra.core.runners.pebble.PebbleUtils;
-import io.kestra.core.services.LogService;
+import io.kestra.core.services.ExecutionLogService;
 import io.kestra.core.utils.ListUtils;
 import io.kestra.core.utils.RetryUtils;
 import io.micronaut.context.annotation.Requires;
@@ -23,14 +23,11 @@ import java.util.Map;
 @Requires(property = "kestra.repository.type")
 public class ErrorLogsFunction  implements Function {
     @Inject
-    private LogService logService;
+    private ExecutionLogService logService;
 
     @Inject
     private PebbleUtils pebbleUtils;
 
-    @Inject
-    private RetryUtils retryUtils;
-
     @Override
     public List<String> getArgumentNames() {
         return Collections.emptyList();
@@ -46,7 +43,7 @@ public class ErrorLogsFunction  implements Function {
         Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
         Map<String, String> execution = (Map<String, String>) context.getVariable("execution");
 
-        RetryUtils.Instance<List<LogEntry>, Throwable> retry = retryUtils.of(Exponential.builder()
+        RetryUtils.Instance<List<LogEntry>, Throwable> retry = RetryUtils.of(Exponential.builder()
             .delayFactor(2.0)
             .interval(Duration.ofMillis(100))
             .maxInterval(Duration.ofSeconds(1))

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

@@ -1,22 +1,30 @@
 package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.runners.LocalPath;
+import io.kestra.core.storages.Namespace;
+import io.kestra.core.storages.NamespaceFile;
 import io.kestra.core.storages.StorageContext;
 import io.pebbletemplates.pebble.template.EvaluationContext;
 import jakarta.inject.Singleton;
 
 import java.io.IOException;
 import java.net.URI;
+import java.nio.file.Path;
+import java.util.Map;
 
 @Singleton
 public class FileExistsFunction extends AbstractFileFunction {
     private static final String ERROR_MESSAGE = "The 'fileExists' function expects an argument 'path' that is a path to the internal storage URI.";
 
     @Override
-    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
+    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
         return switch (path.getScheme()) {
             case StorageContext.KESTRA_SCHEME -> storageInterface.exists(tenantId, namespace, path);
             case LocalPath.FILE_SCHEME -> localPathFactory.createLocalPath().exists(path);
+            case Namespace.NAMESPACE_FILE_SCHEME  -> {
+                Namespace namespaceStorage = namespaceFactory.of(tenantId, namespace, storageInterface);
+                yield namespaceStorage.exists(NamespaceFile.normalize(Path.of(path.getPath()), true));
+            }
             default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
         };
     }

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

@@ -2,19 +2,23 @@ package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.runners.LocalPath;
 import io.kestra.core.storages.FileAttributes;
+import io.kestra.core.storages.Namespace;
+import io.kestra.core.storages.NamespaceFile;
 import io.kestra.core.storages.StorageContext;
 import io.pebbletemplates.pebble.template.EvaluationContext;
 import jakarta.inject.Singleton;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.file.Path;
 import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Map;
 
 @Singleton
 public class FileSizeFunction extends AbstractFileFunction {
     private static final String ERROR_MESSAGE = "The 'fileSize' function expects an argument 'path' that is a path to the internal storage URI.";
 
     @Override
-    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
+    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
         return switch (path.getScheme()) {
             case StorageContext.KESTRA_SCHEME -> {
                 FileAttributes fileAttributes = storageInterface.getAttributes(tenantId, namespace, path);
@@ -24,6 +28,12 @@ public class FileSizeFunction extends AbstractFileFunction {
                 BasicFileAttributes fileAttributes = localPathFactory.createLocalPath().getAttributes(path);
                 yield fileAttributes.size();
             }
+            case Namespace.NAMESPACE_FILE_SCHEME  -> {
+                FileAttributes fileAttributes = namespaceFactory
+                    .of(tenantId, namespace, storageInterface)
+                    .getFileMetadata(NamespaceFile.normalize(Path.of(path.getPath()), true));
+                yield fileAttributes.getSize();
+            }
             default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
         };
     }

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

@@ -1,19 +1,24 @@
 package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.runners.LocalPath;
+import io.kestra.core.storages.FileAttributes;
+import io.kestra.core.storages.Namespace;
+import io.kestra.core.storages.NamespaceFile;
 import io.kestra.core.storages.StorageContext;
 import io.pebbletemplates.pebble.template.EvaluationContext;
 import jakarta.inject.Singleton;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
+import java.nio.file.Path;
+import java.util.Map;
 
 @Singleton
 public class IsFileEmptyFunction extends AbstractFileFunction {
     private static final String ERROR_MESSAGE = "The 'isFileEmpty' function expects an argument 'path' that is a path to a namespace file or an internal storage URI.";
 
     @Override
-    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
+    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
         return switch (path.getScheme()) {
             case StorageContext.KESTRA_SCHEME -> {
                 try (InputStream inputStream = storageInterface.get(tenantId, namespace, path)) {
@@ -27,6 +32,12 @@ public class IsFileEmptyFunction extends AbstractFileFunction {
                     yield inputStream.read(buffer, 0, 1) <= 0;
                 }
             }
+            case Namespace.NAMESPACE_FILE_SCHEME -> {
+                FileAttributes fileAttributes = namespaceFactory
+                    .of(tenantId, namespace, storageInterface)
+                    .getFileMetadata(NamespaceFile.normalize(Path.of(path.getPath()), true));
+                yield fileAttributes.getSize() <= 0;
+            }
             default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
         };
     }
@@ -35,4 +46,4 @@ public class IsFileEmptyFunction extends AbstractFileFunction {
     protected String getErrorMessage() {
         return ERROR_MESSAGE;
     }
-}
+}

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

@@ -1,20 +1,37 @@
 package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.runners.LocalPath;
+import io.kestra.core.storages.Namespace;
+import io.kestra.core.storages.NamespaceFile;
 import io.kestra.core.storages.StorageContext;
 import io.pebbletemplates.pebble.template.EvaluationContext;
 import jakarta.inject.Singleton;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
 
 @Singleton
 public class ReadFileFunction extends AbstractFileFunction {
+    public static final String VERSION = "version";
+
     private static final String ERROR_MESSAGE = "The 'read' function expects an argument 'path' that is a path to a namespace file or an internal storage URI.";
 
     @Override
-    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
+    public List<String> getArgumentNames() {
+        return Stream.concat(
+            super.getArgumentNames().stream(),
+            Stream.of(VERSION)
+        ).toList();
+    }
+
+    @Override
+    protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
         return switch (path.getScheme()) {
             case StorageContext.KESTRA_SCHEME -> {
                 try (InputStream inputStream = storageInterface.get(tenantId, namespace, path)) {
@@ -26,12 +43,30 @@ public class ReadFileFunction extends AbstractFileFunction {
                     yield new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
                 }
             }
+            case Namespace.NAMESPACE_FILE_SCHEME -> {
+                try (InputStream inputStream = contentInputStream(path, namespace, tenantId, args)) {
+                    yield new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
+                }
+            }
             default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
         };
     }
 
+    private InputStream contentInputStream(URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
+        Namespace namespaceStorage = namespaceFactory.of(tenantId, namespace, storageInterface);
+
+        if (args.containsKey(VERSION)) {
+            return namespaceStorage.getFileContent(
+                    NamespaceFile.normalize(Path.of(path.getPath()), true),
+                    Integer.parseInt(args.get(VERSION).toString())
+                );
+        }
+
+        return namespaceStorage.getFileContent(NamespaceFile.normalize(Path.of(path.getPath()), true));
+    }
+
     @Override
     protected String getErrorMessage() {
         return ERROR_MESSAGE;
     }
-}
+}

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

@@ -9,6 +9,7 @@ import io.kestra.core.secret.SecretNotFoundException;
 import io.kestra.core.secret.SecretService;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.services.FlowService;
+import io.kestra.core.services.NamespaceService;
 import io.pebbletemplates.pebble.error.PebbleException;
 import io.pebbletemplates.pebble.extension.Function;
 import io.pebbletemplates.pebble.template.EvaluationContext;
@@ -36,7 +37,7 @@ public class SecretFunction implements Function {
     private SecretService secretService;
 
     @Inject
-    private FlowService flowService;
+    private NamespaceService namespaceService;
 
     @Override
     public List<String> getArgumentNames() {
@@ -56,7 +57,7 @@ public class SecretFunction implements Function {
         if (namespace == null) {
             namespace = flowNamespace;
         } else {
-            flowService.checkAllowedNamespace(flowTenantId, namespace, flowTenantId, flowNamespace);
+            namespaceService.checkAllowedNamespace(flowTenantId, namespace, flowTenantId, flowNamespace);
         }
 
         try {

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

@@ -26,7 +26,14 @@ public class ListOrMapOfLabelDeserializer extends JsonDeserializer<List<Label>>
         else if (p.hasToken(JsonToken.START_ARRAY)) {
             // deserialize as list
             List<Map<String, String>> ret = ctxt.readValue(p, List.class);
-            return ret.stream().map(map -> new Label(map.get("key"), map.get("value"))).toList();
+            return ret.stream().map(map -> {
+                Object value = map.get("value");
+                if (isAllowedType(value)) {
+                    return new Label(map.get("key"), String.valueOf(value));
+                } else {
+                    throw new IllegalArgumentException("Unsupported type for key: " + map.get("key") + ", value: " + value);
+                }
+            }).toList();
         }
         else if (p.hasToken(JsonToken.START_OBJECT)) {
             // deserialize as map

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

@@ -81,7 +81,24 @@ public final class YamlParser {
             throw toConstraintViolationException(input, resource, e);
         }
     }
-
+    private static String formatYamlErrorMessage(String originalMessage, JsonProcessingException e) {
+        StringBuilder friendlyMessage = new StringBuilder();
+        if (originalMessage.contains("Expected a field name")) {
+            friendlyMessage.append("YAML syntax error: Invalid structure. Check indentation and ensure all fields are properly formatted.");
+        } else if (originalMessage.contains("MappingStartEvent")) {
+            friendlyMessage.append("YAML syntax error: Unexpected mapping start. Verify that scalar values are properly quoted if needed.");
+        } else if (originalMessage.contains("Scalar value")) {
+            friendlyMessage.append("YAML syntax error: Expected a simple value but found complex structure. Check for unquoted special characters.");
+        } else {
+            friendlyMessage.append("YAML parsing error: ").append(originalMessage.replaceAll("org\\.yaml\\.snakeyaml.*", "").trim());
+        }
+        if (e.getLocation() != null) {
+            int line = e.getLocation().getLineNr();
+            friendlyMessage.append(String.format(" (at line %d)", line));
+        }
+        // Return a generic but cleaner message for other YAML errors
+        return friendlyMessage.toString();
+    }
     @SuppressWarnings("unchecked")
     public static <T> ConstraintViolationException toConstraintViolationException(T target, String resource, JsonProcessingException e) {
         if (e.getCause() instanceof ConstraintViolationException constraintViolationException) {
@@ -121,11 +138,12 @@ public final class YamlParser {
                     )
                 ));
         } else {
+            String userFriendlyMessage = formatYamlErrorMessage(e.getMessage(), e);
             return new ConstraintViolationException(
-                "Illegal " + resource + " source: " + e.getMessage(),
+                "Illegal " + resource + " source: " + userFriendlyMessage,
                 Collections.singleton(
                     ManualConstraintViolation.of(
-                        e.getCause() == null ? e.getMessage() : e.getMessage() + "\nCaused by: " + e.getCause().getMessage(),
+                        userFriendlyMessage,
                         target,
                         (Class<T>) target.getClass(),
                         "yaml",
@@ -136,4 +154,3 @@ public final class YamlParser {
         }
     }
 }
-

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

@@ -4,7 +4,6 @@ import com.cronutils.utils.VisibleForTesting;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
-import io.kestra.core.models.conditions.ScheduleCondition;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowInterface;
@@ -65,16 +64,6 @@ public class ConditionService {
         return this.valid(flow, conditions, conditionContext);
     }
 
-    /**
-     * Check that all conditions are valid.
-     * Warning, this method throws if a condition cannot be evaluated.
-     */
-    public boolean isValid(List<ScheduleCondition> conditions, ConditionContext conditionContext) throws InternalException {
-        return conditions
-            .stream()
-            .allMatch(throwPredicate(condition -> condition.test(conditionContext)));
-    }
-
     /**
      * Check that all conditions are valid.
      * Warning, this method throws if a condition cannot be evaluated.

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

@@ -2,12 +2,15 @@ package io.kestra.core.services;
 
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.repositories.LogRepositoryInterface;
+import io.micronaut.data.model.Pageable;
+import io.micronaut.data.model.Sort;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.slf4j.event.Level;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -17,9 +20,42 @@ import java.util.stream.Stream;
  */
 @Singleton
 public class ExecutionLogService {
+   
+    private final LogRepositoryInterface logRepository;
+    
     @Inject
-    private LogRepositoryInterface logRepository;
+    public ExecutionLogService(LogRepositoryInterface logRepository) {
+        this.logRepository = logRepository;
+    }
 
+    /**
+     * Purges log entries matching the given criteria.
+     *
+     * @param tenantId    the tenant identifier
+     * @param namespace   the namespace of the flow
+     * @param flowId      the flow identifier
+     * @param executionId the execution identifier
+     * @param logLevels   the list of log levels to delete
+     * @param startDate   the start of the date range
+     * @param endDate     the end of the date range.
+     * @return the number of log entries deleted
+     */
+    public int purge(String tenantId, String namespace, String flowId, String executionId, List<Level> logLevels, ZonedDateTime startDate, ZonedDateTime endDate) {
+        return logRepository.deleteByQuery(tenantId, namespace, flowId, executionId, logLevels, startDate, endDate);
+    }
+
+
+    /**
+     * Fetches the error logs of an execution.
+     * <p>
+     * This method limits the results to the first 25 error logs, ordered by timestamp asc.
+     *
+     * @return the log entries
+     */
+    public List<LogEntry> errorLogs(String tenantId, String executionId) {
+        return logRepository.findByExecutionId(tenantId, executionId, Level.ERROR, Pageable.from(1, 25, Sort.of(Sort.Order.asc("timestamp"))));
+    }
+    
     public InputStream getExecutionLogsAsStream(String tenantId,
                                                 String executionId,
                                                 Level minLevel,

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

@@ -2,8 +2,10 @@ package io.kestra.core.services;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import io.kestra.core.exceptions.FlowProcessingException;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.*;
+import io.kestra.core.models.flows.check.Check;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.topologies.FlowTopology;
 import io.kestra.core.models.triggers.AbstractTrigger;
@@ -12,10 +14,13 @@ import io.kestra.core.models.validations.ValidateConstraintViolation;
 import io.kestra.core.plugins.PluginRegistry;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.FlowTopologyRepositoryInterface;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.utils.ListUtils;
 import io.kestra.plugin.core.flow.Pause;
 import jakarta.inject.Inject;
+import jakarta.inject.Provider;
 import jakarta.inject.Singleton;
 import jakarta.validation.ConstraintViolationException;
 import lombok.extern.slf4j.Slf4j;
@@ -54,6 +59,9 @@ public class FlowService {
     @Inject
     Optional<FlowTopologyRepositoryInterface> flowTopologyRepository;
 
+    @Inject
+    Provider<RunContextFactory> runContextFactory; // Lazy init: avoid circular dependency error.
+
     /**
      * Validates and creates the given flow.
      * <p>
@@ -84,6 +92,57 @@ public class FlowService {
         return flowRepository
             .orElseThrow(() -> new IllegalStateException("Cannot perform operation on flow. Cause: No FlowRepository"));
     }
+    private static String formatValidationError(String message) {
+        if (message.startsWith("Illegal flow source:")) {
+            // Already formatted by YamlParser, return as-is
+            return message;
+        }
+        // For other validation errors, provide context
+        return "Validation error: " + message;
+    }
+    /**
+     * Evaluates all checks defined in the given flow using the provided inputs.
+     * <p>
+     * Each check's {@link Check#getCondition()} is evaluated in the context of the flow.
+     * If a condition evaluates to {@code false} or fails to evaluate due to a
+     * variable error, the corresponding {@link Check} is added to the returned list.
+     * </p>
+     *
+     * @param flow   the flow containing the checks to evaluate
+     * @param inputs the input values used when evaluating the conditions
+     * @return a list of checks whose conditions evaluated to {@code false} or failed to evaluate
+     */
+    public List<Check> getFailedChecks(Flow flow, Map<String, Object> inputs) {
+        if (!ListUtils.isEmpty(flow.getChecks())) {
+            RunContext runContext = runContextFactory.get().of(flow, Map.of("inputs", inputs));
+            List<Check> falseConditions = new ArrayList<>();
+            for (Check check : flow.getChecks()) {
+                try {
+                    boolean result = Boolean.TRUE.equals(runContext.renderTyped(check.getCondition()));
+                    if (!result) {
+                        falseConditions.add(check);
+                    }
+                } catch (IllegalVariableEvaluationException e) {
+                    log.debug("[tenant: {}] [namespace: {}] [flow: {}] Failed to evaluate check condition. Cause.: {}",
+                        flow.getTenantId(),
+                        flow.getNamespace(),
+                        flow.getId(),
+                        e.getMessage(),
+                        e
+                    );
+                    falseConditions.add(Check
+                        .builder()
+                            .message("Failed to evaluate check condition. Cause: " + e.getMessage())
+                            .behavior(Check.Behavior.BLOCK_EXECUTION)
+                            .style(Check.Style.ERROR)
+                        .build()
+                    );
+                }
+            }
+            return falseConditions;
+        }
+        return List.of();
+    }
 
     /**
      * Validates the given flow source.
@@ -122,10 +181,12 @@ public class FlowService {
                     modelValidator.validate(pluginDefaultService.injectAllDefaults(flow, false));
 
                 } catch (ConstraintViolationException e) {
-                    validateConstraintViolationBuilder.constraints(e.getMessage());
+                    String friendlyMessage = formatValidationError(e.getMessage());
+                    validateConstraintViolationBuilder.constraints(friendlyMessage);
                 } catch (FlowProcessingException e) {
-                    if (e.getCause() instanceof ConstraintViolationException) {
-                        validateConstraintViolationBuilder.constraints(e.getMessage());
+                    if (e.getCause() instanceof ConstraintViolationException cve) {
+                        String friendlyMessage = formatValidationError(cve.getMessage());
+                        validateConstraintViolationBuilder.constraints(friendlyMessage);
                     } else {
                         Throwable cause = e.getCause() != null ? e.getCause() : e;
                         validateConstraintViolationBuilder.constraints("Unable to validate the flow: " + cause.getMessage());
@@ -456,50 +517,6 @@ public class FlowService {
         return flowRepository.get().delete(flow);
     }
 
-    /**
-     * Return true if the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
-     * As namespace restriction is an EE feature, this will always return true in OSS.
-     */
-    public boolean isAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
-        return true;
-    }
-
-    /**
-     * Check that the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
-     * If not, throw an IllegalArgumentException.
-     */
-    public void checkAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
-        if (!isAllowedNamespace(tenant, namespace, fromTenant, fromNamespace)) {
-            throw new IllegalArgumentException("Namespace " + namespace + " is not allowed.");
-        }
-    }
-
-    /**
-     * Return true if the namespace is allowed from all the namespace in the 'fromTenant' tenant.
-     * As namespace restriction is an EE feature, this will always return true in OSS.
-     */
-    public boolean areAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
-        return true;
-    }
-
-    /**
-     * Check that the namespace is allowed from all the namespace in the 'fromTenant' tenant.
-     * If not, throw an IllegalArgumentException.
-     */
-    public void checkAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
-        if (!areAllowedAllNamespaces(tenant, fromTenant, fromNamespace)) {
-            throw new IllegalArgumentException("All namespaces are not allowed, you should either filter on a namespace or configure all namespaces to allow your namespace.");
-        }
-    }
-
-    /**
-     * Return true if require existing namespace is enabled and the namespace didn't already exist.
-     * As namespace management is an EE feature, this will always return false in OSS.
-     */
-    public boolean requireExistingNamespace(String tenant, String namespace) {
-        return false;
-    }
-
     /**
      * Gets the executable flow for the given namespace, id, and revision.
      * Warning: this method bypasses ACL so someone with only execution right can create a flow execution
@@ -571,4 +588,4 @@ public class FlowService {
     private IllegalStateException noRepositoryException() {
         return new IllegalStateException("No repository found. Make sure the `kestra.repository.type` property is set.");
     }
-}
+}

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

@@ -20,9 +20,6 @@ public class KVStoreService {
     @Inject
     private StorageInterface storageInterface;
 
-    @Inject
-    private FlowService flowService;
-
     @Inject
     private NamespaceService namespaceService;
 
@@ -38,7 +35,7 @@ public class KVStoreService {
         boolean isNotSameNamespace = fromNamespace != null && !namespace.equals(fromNamespace);
         if (isNotSameNamespace && isNotParentNamespace(namespace, fromNamespace)) {
             try {
-                flowService.checkAllowedNamespace(tenant, namespace, tenant, fromNamespace);
+                namespaceService.checkAllowedNamespace(tenant, namespace, tenant, fromNamespace);
             } catch (IllegalArgumentException e) {
                 throw new KVStoreException(String.format(
                     "Cannot access the KV store. Access to '%s' namespace is not allowed from '%s'.", namespace, fromNamespace)

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

@@ -1,5 +1,6 @@
 package io.kestra.core.services;
 
+import io.kestra.core.exceptions.ResourceAccessDeniedException;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.utils.NamespaceUtils;
 import jakarta.inject.Inject;
@@ -39,4 +40,52 @@ public class NamespaceService {
         }
         return false;
     }
+
+    /**
+     * Return true if require existing namespace is enabled and the namespace didn't already exist.
+     * As namespace management is an EE feature, this will always return false in OSS.
+     */
+    public boolean requireExistingNamespace(String tenant, String namespace) {
+        return false;
+    }
+
+    /**
+     * Return true if the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
+     * As namespace restriction is an EE feature, this will always return true in OSS.
+     */
+    public boolean isAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
+        return true;
+    }
+
+    /**
+     * Check that the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
+     * If not, throw a ResourceAccessDeniedException.
+     *
+     * @throws ResourceAccessDeniedException if the namespace is not allowed.
+     */
+    public void checkAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
+        if (!isAllowedNamespace(tenant, namespace, fromTenant, fromNamespace)) {
+            throw new ResourceAccessDeniedException("Namespace " + namespace + " is not allowed.");
+        }
+    }
+
+    /**
+     * Return true if the namespace is allowed from all the namespace in the 'fromTenant' tenant.
+     * As namespace restriction is an EE feature, this will always return true in OSS.
+     */
+    public boolean areAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
+        return true;
+    }
+
+    /**
+     * Check that the namespace is allowed from all the namespace in the 'fromTenant' tenant.
+     * If not, throw a ResourceAccessDeniedException.
+     *
+     * @throws ResourceAccessDeniedException if all namespaces all aren't allowed.
+     */
+    public void checkAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
+        if (!areAllowedAllNamespaces(tenant, fromTenant, fromNamespace)) {
+            throw new ResourceAccessDeniedException("All namespaces are not allowed, you should either filter on a namespace or configure all namespaces to allow your namespace.");
+        }
+    }
 }

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

@@ -23,6 +23,7 @@ import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.runners.RunContextLogger;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.serializers.YamlParser;
+import io.kestra.core.utils.Logs;
 import io.kestra.core.utils.MapUtils;
 import io.kestra.plugin.core.flow.Template;
 import io.micronaut.context.annotation.Value;
@@ -30,7 +31,6 @@ import io.micronaut.core.annotation.Nullable;
 import jakarta.annotation.PostConstruct;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
-import jakarta.inject.Provider;
 import jakarta.inject.Singleton;
 import jakarta.validation.ConstraintViolationException;
 import lombok.extern.slf4j.Slf4j;
@@ -82,10 +82,7 @@ public class PluginDefaultService {
 
     @Inject
     protected PluginRegistry pluginRegistry;
-
-    @Inject
-    protected Provider<LogService> logService; // lazy-init
-
+    
     @Value("{kestra.templates.enabled:false}")
     private boolean templatesEnabled;
 
@@ -255,7 +252,7 @@ public class PluginDefaultService {
         if (source == null) {
             // This should never happen
             String error = "Cannot apply plugin defaults. Cause: flow has no defined source.";
-            logService.get().logExecution(flow, log, Level.ERROR, error);
+            Logs.logExecution(flow, log, Level.ERROR, error);
             throw new IllegalArgumentException(error);
         }
 
@@ -311,7 +308,7 @@ public class PluginDefaultService {
             result = parseFlowWithAllDefaults(flow.getTenantId(), flow.getNamespace(), flow.getRevision(), flow.isDeleted(), source, true, false);
         } catch (Exception e) {
             if (safe) {
-                logService.get().logExecution(flow, log, Level.ERROR, "Failed to read flow.", e);
+                Logs.logExecution(flow, log, Level.ERROR, "Failed to read flow.", e);
                 result = FlowWithException.from(flow, e);
 
                 // deleted is not part of the original 'source'

core/src/main/java/io/kestra/core/storages/InternalNamespace.java

@@ -1,18 +1,27 @@
 package io.kestra.core.storages;
 
+import io.kestra.core.models.FetchVersion;
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
+import io.kestra.core.repositories.ArrayListTotal;
+import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
+import io.micronaut.data.model.Pageable;
 import jakarta.annotation.Nullable;
+import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import static io.kestra.core.utils.Rethrow.throwFunction;
 
 /**
  * The default {@link Namespace} implementation.
@@ -28,6 +37,7 @@ public class InternalNamespace implements Namespace {
     private final String namespace;
     private final String tenant;
     private final StorageInterface storage;
+    private final NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository;
     private final Logger logger;
 
     /**
@@ -36,8 +46,8 @@ public class InternalNamespace implements Namespace {
      * @param namespace The namespace
      * @param storage   The storage.
      */
-    public InternalNamespace(@Nullable final String tenant, final String namespace, final StorageInterface storage) {
-        this(LOG, tenant, namespace, storage);
+    public InternalNamespace(@Nullable final String tenant, final String namespace, final StorageInterface storage, final NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository) {
+        this(LOG, tenant, namespace, storage, namespaceFileMetadataRepository);
     }
 
     /**
@@ -48,13 +58,27 @@ public class InternalNamespace implements Namespace {
      * @param tenant    The tenant.
      * @param storage   The storage.
      */
-    public InternalNamespace(final Logger logger, @Nullable final String tenant, final String namespace, final StorageInterface storage) {
+    public InternalNamespace(final Logger logger, @Nullable final String tenant, final String namespace, final StorageInterface storage, final NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepositoryInterface) {
         this.logger = Objects.requireNonNull(logger, "logger cannot be null");
         this.namespace = Objects.requireNonNull(namespace, "namespace cannot be null");
         this.storage = Objects.requireNonNull(storage, "storage cannot be null");
+        this.namespaceFileMetadataRepository = Objects.requireNonNull(namespaceFileMetadataRepositoryInterface, "namespaceFileMetadataRepository cannot be null");
         this.tenant = tenant;
     }
 
+    @Override
+    public ArrayListTotal<NamespaceFile> find(Pageable pageable, List<QueryFilter> filters, boolean allowDeleted, FetchVersion fetchVersion) {
+        return namespaceFileMetadataRepository.find(
+            pageable,
+            tenant,
+            Stream.concat(filters.stream(), Stream.of(
+                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build()
+            )).toList(),
+            allowDeleted,
+            fetchVersion
+        ).map(throwFunction(NamespaceFile::fromMetadata));
+    }
+
     /**
      * {@inheritDoc}
      **/
@@ -73,35 +97,106 @@ public class InternalNamespace implements Namespace {
      **/
     @Override
     public List<NamespaceFile> all() throws IOException {
-        return all(false);
+        return all(null);
     }
 
     /**
      * {@inheritDoc}
      **/
     @Override
-    public List<NamespaceFile> all(final boolean includeDirectories) throws IOException {
-        return all(null, includeDirectories);
+    public List<NamespaceFile> all(final String containing, boolean includeDirectories) throws IOException {
+        List<NamespaceFileMetadata> namespaceFilesMetadata = namespaceFileMetadataRepository.find(Pageable.UNPAGED, tenant, Stream.concat(
+            Stream.of(QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build()),
+            Optional.ofNullable(containing).flatMap(p -> {
+                if (p.equals("/")) {
+                    return Optional.empty();
+                }
+
+                return Optional.of(QueryFilter.builder().field(QueryFilter.Field.QUERY).operation(QueryFilter.Op.EQUALS).value(p).build());
+            }).stream()
+        ).toList(), false);
+
+        if (!includeDirectories) {
+            namespaceFilesMetadata = namespaceFilesMetadata.stream().filter(nsFileMetadata -> !nsFileMetadata.isDirectory()).toList();
+        }
+
+        return namespaceFilesMetadata.stream().filter(nsFileMetadata -> !nsFileMetadata.getPath().equals("/")).map(nsFileMetadata -> NamespaceFile.of(namespace, Path.of(nsFileMetadata.getPath()), nsFileMetadata.getVersion())).toList();
     }
 
     /**
      * {@inheritDoc}
      **/
     @Override
-    public List<NamespaceFile> all(final String prefix, final boolean includeDirectories) throws IOException {
-        URI namespacePrefix = URI.create(NamespaceFile.of(namespace, Optional.ofNullable(prefix).map(Path::of).orElse(null)).storagePath().toString().replace("\\","/") + "/");
-        return storage.allByPrefix(tenant, namespace, namespacePrefix, includeDirectories)
-            .stream()
-            .map(uri -> new NamespaceFile(relativize(uri), uri, namespace))
-            .toList();
+    public List<NamespaceFileMetadata> children(String parentPath, boolean recursive) throws IOException {
+        final String normalizedParentPath = NamespaceFile.normalize(Path.of(parentPath), true).toString();
+
+        return namespaceFileMetadataRepository.find(Pageable.UNPAGED, tenant, List.of(
+            QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
+            QueryFilter.builder()
+                .field(QueryFilter.Field.PARENT_PATH)
+                .operation(recursive ? QueryFilter.Op.STARTS_WITH : QueryFilter.Op.EQUALS)
+                .value(normalizedParentPath.endsWith("/") ? normalizedParentPath : normalizedParentPath + "/")
+                .build()
+        ), false);
+    }
+
+    @Override
+    public List<Pair<NamespaceFile, NamespaceFile>> move(Path source, Path target) throws Exception {
+        final Path normalizedSource = NamespaceFile.normalize(source, true);
+        final Path normalizedTarget = NamespaceFile.normalize(target, true);
+
+        if (findByPath(normalizedTarget).isPresent()) {
+            throw new IOException(String.format(
+                "File '%s' already exists in namespace '%s'.",
+                normalizedTarget,
+                namespace
+            ));
+        }
+
+        ArrayListTotal<NamespaceFileMetadata> beforeRename = namespaceFileMetadataRepository.find(Pageable.UNPAGED, tenant, List.of(
+            QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
+            QueryFilter.builder().field(QueryFilter.Field.PATH).operation(QueryFilter.Op.IN).value(List.of(normalizedSource.toString(), normalizedSource + "/")).build()
+        ), true, FetchVersion.ALL);
+        beforeRename.sort(Comparator.comparing(NamespaceFileMetadata::getVersion));
+        ArrayListTotal<NamespaceFileMetadata> afterRename = beforeRename
+            .map(nsFileMetadata -> {
+                String newPath;
+                if (nsFileMetadata.isDirectory()) {
+                    newPath = normalizedTarget.toString().endsWith("/") ? normalizedTarget.toString() : normalizedTarget + "/";
+                } else {
+                    newPath = normalizedTarget.toString();
+                }
+
+                return nsFileMetadata.toBuilder().path(newPath).build();
+            });
+
+        return afterRename.map(throwFunction(nsFileMetadata -> {
+            NamespaceFile beforeNamespaceFile = NamespaceFile.of(namespace, normalizedSource, nsFileMetadata.getVersion());
+            Path namespaceFilePath = beforeNamespaceFile.storagePath();
+            NamespaceFile afterNamespaceFile;
+            if (nsFileMetadata.isDirectory()) {
+                afterNamespaceFile = this.createDirectory(Path.of(nsFileMetadata.getPath()));
+            } else {
+                try (InputStream oldContent = storage.get(tenant, namespace, namespaceFilePath.toUri())) {
+                    afterNamespaceFile = this.putFile(Path.of(nsFileMetadata.getPath()), oldContent, Conflicts.OVERWRITE).getFirst();
+                }
+            }
+
+            this.purge(NamespaceFile.of(namespace, normalizedSource, nsFileMetadata.getVersion()));
+            return Pair.of(beforeNamespaceFile, afterNamespaceFile);
+        }));
     }
 
     /**
      * {@inheritDoc}
      **/
     @Override
-    public NamespaceFile get(final Path path) {
-        return NamespaceFile.of(namespace, path);
+    public NamespaceFile get(Path path) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        int version = findByPath(normalizedPath).map(NamespaceFileMetadata::getVersion).orElse(1);
+
+        return NamespaceFile.of(namespace, normalizedPath, version);
     }
 
     public Path relativize(final URI uri) {
@@ -122,90 +217,225 @@ public class InternalNamespace implements Namespace {
      * {@inheritDoc}
      **/
     @Override
-    public InputStream getFileContent(final Path path) throws IOException {
-        Path namespaceFilePath = NamespaceFile.of(namespace, path).storagePath();
+    public InputStream getFileContent(Path path, @Nullable Integer version) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        // Throw if file not found OR if it's deleted
+        NamespaceFileMetadata namespaceFileMetadata = findByPath(normalizedPath, version).orElseThrow(() -> fileNotFound(normalizedPath, version));
+
+        Path namespaceFilePath = NamespaceFile.of(namespace, normalizedPath, namespaceFileMetadata.getVersion()).storagePath();
         return storage.get(tenant, namespace, namespaceFilePath.toUri());
     }
 
+    @Override
+    public FileAttributes getFileMetadata(Path path) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        return findByPath(normalizedPath).map(NamespaceFileAttributes::new).orElseThrow(() -> fileNotFound(normalizedPath, null));
+    }
+
+    private FileNotFoundException fileNotFound(Path path, @Nullable Integer version) {
+        return new FileNotFoundException(Optional.ofNullable(version).map(v -> "Version " + v + " of file").orElse("File") + " '" + path + "' was not found in namespace '" + namespace + "'.");
+    }
+
+    private Optional<NamespaceFileMetadata> findByPath(Path path, boolean allowDeleted, @Nullable Integer version) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        if (version != null) {
+            return namespaceFileMetadataRepository.find(Pageable.from(1, 1), tenant, List.of(
+                QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
+                QueryFilter.builder().field(QueryFilter.Field.PATH).operation(QueryFilter.Op.EQUALS).value(normalizedPath.toString()).build(),
+                QueryFilter.builder().field(QueryFilter.Field.VERSION).operation(QueryFilter.Op.EQUALS).value(version).build()
+            ), allowDeleted, FetchVersion.ALL).stream().findFirst();
+        }
+        return namespaceFileMetadataRepository.findByPath(tenant, namespace, normalizedPath.toString())
+            .filter(namespaceFileMetadata -> allowDeleted || !namespaceFileMetadata.isDeleted());
+    }
+
+    private Optional<NamespaceFileMetadata> findByPath(Path path, boolean allowDeleted) throws IOException {
+        return findByPath(path, allowDeleted, null);
+    }
+
+    private Optional<NamespaceFileMetadata> findByPath(Path path, @Nullable Integer version) throws IOException {
+        return findByPath(path, false, version);
+    }
+
+    private Optional<NamespaceFileMetadata> findByPath(Path path) throws IOException {
+        return findByPath(path, null);
+    }
+
+    @Override
+    public boolean exists(Path path) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        return findByPath(normalizedPath).isPresent();
+    }
+
     /**
      * {@inheritDoc}
      **/
     @Override
-    public NamespaceFile putFile(final Path path, final InputStream content, final Conflicts onAlreadyExist) throws IOException, URISyntaxException {
-        Path namespaceFilesPrefix = NamespaceFile.of(namespace, path).storagePath();
+    public List<NamespaceFile> putFile(final Path path, final InputStream content, final Conflicts onAlreadyExist) throws IOException, URISyntaxException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        Optional<NamespaceFileMetadata> inRepository = findByPath(normalizedPath, true);
+        int currentVersion = inRepository.map(NamespaceFileMetadata::getVersion).orElse(0);
+        NamespaceFile namespaceFile = NamespaceFile.of(namespace, normalizedPath, currentVersion + 1);
+        Path storagePath = namespaceFile.storagePath();
         // Remove Windows letter
-        URI cleanUri = new URI(namespaceFilesPrefix.toUri().toString().replaceFirst("^file:///[a-zA-Z]:", ""));
-        final boolean exists = storage.exists(tenant, namespace, cleanUri);
+        URI cleanUri = new URI(storagePath.toUri().toString().replaceFirst("^file:///[a-zA-Z]:", ""));
 
-        return switch (onAlreadyExist) {
-            case OVERWRITE -> {
-                URI uri = storage.put(tenant, namespace, cleanUri, content);
-                NamespaceFile namespaceFile = new NamespaceFile(relativize(uri), uri, namespace);
-                if (exists) {
-                    logger.debug(String.format(
-                        "File '%s' overwritten into namespace '%s'.",
-                        path,
-                        namespace
-                    ));
-                } else {
-                    logger.debug(String.format(
-                        "File '%s' added to namespace '%s'.",
-                        path,
-                        namespace
-                    ));
-                }
-                yield namespaceFile;
+        List<NamespaceFile> createdFiles = new ArrayList<>();
+        if (inRepository.isEmpty()) {
+            storage.put(tenant, namespace, cleanUri, content);
+
+            createdFiles.addAll(mkDirs(normalizedPath.toString()));
+
+            namespaceFileMetadataRepository.save(
+                NamespaceFileMetadata.builder()
+                    .tenantId(tenant)
+                    .namespace(namespace)
+                    .path(normalizedPath.toString())
+                    .size(storage.getAttributes(tenant, namespace, cleanUri).getSize())
+                    .build()
+            );
+
+            logger.debug(String.format(
+                "File '%s' added to namespace '%s'.",
+                normalizedPath,
+                namespace
+            ));
+
+            createdFiles.add(namespaceFile);
+        } else if (onAlreadyExist == Conflicts.OVERWRITE || inRepository.get().isDeleted()) {
+            storage.put(tenant, namespace, cleanUri, content);
+
+            createdFiles.addAll(mkDirs(normalizedPath.toString()));
+
+            namespaceFileMetadataRepository.save(
+                inRepository.get().toBuilder().size(storage.getAttributes(tenant, namespace, cleanUri).getSize()).deleted(false).build()
+            );
+
+            if (inRepository.get().isDeleted()) {
+                logger.debug(String.format(
+                    "File '%s' added to namespace '%s'.",
+                    normalizedPath,
+                    namespace
+                ));
+            } else {
+                logger.debug(String.format(
+                    "File '%s' overwritten into namespace '%s'.",
+                    normalizedPath,
+                    namespace
+                ));
             }
-            case ERROR -> {
-                if (!exists) {
-                    URI uri = storage.put(tenant, namespace, namespaceFilesPrefix.toUri(), content);
-                    yield new NamespaceFile(relativize(uri), uri, namespace);
-                } else {
-                    throw new IOException(String.format(
-                        "File '%s' already exists in namespace '%s' and conflict is set to %s",
-                        path,
-                        namespace,
-                        Conflicts.ERROR
-                    ));
-                }
+
+            createdFiles.add(namespaceFile);
+        } else {
+            // At this point, the file exists and we have to decide what to do based on the conflict strategy
+            switch (onAlreadyExist) {
+                case ERROR -> throw new IOException(String.format(
+                    "File '%s' already exists in namespace '%s' and conflict is set to %s",
+                    normalizedPath,
+                    namespace,
+                    Conflicts.ERROR
+                ));
+                case SKIP -> logger.debug(String.format(
+                    "File '%s' already exists in namespace '%s' and conflict is set to %s. Skipping.",
+                    normalizedPath,
+                    namespace,
+                    Conflicts.SKIP
+                ));
             }
-            case SKIP -> {
-                if (!exists) {
-                    URI uri = storage.put(tenant, namespace, namespaceFilesPrefix.toUri(), content);
-                    NamespaceFile namespaceFile = new NamespaceFile(relativize(uri), uri, namespace);
-                    logger.debug(String.format(
-                        "File '%s' added to namespace '%s'.",
-                        path,
-                        namespace
-                    ));
-                    yield namespaceFile;
-                } else {
-                    logger.debug(String.format(
-                        "File '%s' already exists in namespace '%s' and conflict is set to %s. Skipping.",
-                        path,
-                        namespace,
-                        Conflicts.SKIP
-                    ));
-                    URI uri = URI.create(StorageContext.KESTRA_PROTOCOL + namespaceFilesPrefix);
-                    yield new NamespaceFile(relativize(uri), uri, namespace);
-                }
-            }
-        };
+        }
+
+        return createdFiles;
+    }
+
+    /**
+     * Make all parent directories for a given path.
+     */
+    private List<NamespaceFile> mkDirs(String path) throws IOException {
+        List<NamespaceFile> createdDirs = new ArrayList<>();
+        Optional<Path> maybeParentPath = Optional.empty();
+        while (
+            (maybeParentPath = Optional.ofNullable(NamespaceFileMetadata.parentPath(maybeParentPath.map(Path::toString).orElse(path))).map(Path::of)).isPresent()
+                && !this.exists(maybeParentPath.get())
+        ) {
+            this.createDirectory(maybeParentPath.get());
+            createdDirs.add(NamespaceFile.of(namespace, maybeParentPath.get().toString().endsWith("/") ? maybeParentPath.get().toString() : maybeParentPath.get() + "/", 1));
+        }
+
+        return createdDirs;
     }
 
     /**
      * {@inheritDoc}
      **/
     @Override
-    public URI createDirectory(Path path) throws IOException {
-        return storage.createDirectory(tenant, namespace, NamespaceFile.of(namespace, path).storagePath().toUri());
+    public NamespaceFile createDirectory(Path path) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        NamespaceFileMetadata nsFileMetadata = namespaceFileMetadataRepository.save(
+            NamespaceFileMetadata.builder()
+                .tenantId(tenant)
+                .namespace(namespace)
+                .path(normalizedPath.toString().endsWith("/") ? normalizedPath.toString() : normalizedPath + "/")
+                .size(0L)
+                .build()
+        );
+        storage.createDirectory(tenant, namespace, NamespaceFile.of(namespace, normalizedPath, 1).storagePath().toUri());
+
+        return NamespaceFile.fromMetadata(nsFileMetadata);
     }
 
     /**
      * {@inheritDoc}
      **/
     @Override
-    public boolean delete(Path path) throws IOException {
-        return storage.delete(tenant, namespace, URI.create(path.toString().replace("\\","/")));
+    public List<NamespaceFile> delete(Path path) throws IOException {
+        final Path normalizedPath = NamespaceFile.normalize(path, true);
+
+        Optional<NamespaceFileMetadata> maybeNamespaceFileMetadata = namespaceFileMetadataRepository.find(Pageable.from(1, 1), tenant, List.of(
+            QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
+            QueryFilter.builder().field(QueryFilter.Field.PATH).operation(QueryFilter.Op.IN).value(List.of(normalizedPath.toString(), normalizedPath + "/")).build()
+        ), false).stream().findFirst();
+
+        List<NamespaceFileMetadata> toDelete = Stream.concat(
+            this.children(normalizedPath.toString(), true).stream().map(NamespaceFileMetadata::toDeleted),
+            maybeNamespaceFileMetadata.map(NamespaceFileMetadata::toDeleted).stream()
+        ).toList();
+
+        toDelete.forEach(namespaceFileMetadataRepository::save);
+
+        return toDelete.stream().map(NamespaceFile::fromMetadata).toList();
+    }
+
+    @Override
+    public boolean purge(NamespaceFile namespaceFile) throws IOException {
+        storage.delete(tenant, namespace, namespaceFile.storagePath().toUri());
+        namespaceFileMetadataRepository.purge(List.of(NamespaceFileMetadata.of(tenant, namespaceFile)));
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Integer purge(List<NamespaceFile> namespaceFiles) throws IOException {
+        Integer purgedMetadataCount = this.namespaceFileMetadataRepository.purge(namespaceFiles.stream().map(namespaceFile -> NamespaceFileMetadata.of(tenant, namespaceFile)).toList());
+
+        long actualDeletedEntries = namespaceFiles.stream()
+            .map(NamespaceFile::storagePath)
+            .map(Path::toUri)
+            .map(throwFunction(uri -> this.storage.delete(tenant, namespace, uri)))
+            .filter(Boolean::booleanValue)
+            .count();
+
+        if (actualDeletedEntries != purgedMetadataCount) {
+            LOG.warn("Namespace Files Metadata purge reported {} deleted entries, but {} values were actually deleted from storage", purgedMetadataCount, actualDeletedEntries);
+        }
+
+        return purgedMetadataCount;
     }
 }

core/src/main/java/io/kestra/core/storages/InternalStorage.java

@@ -1,15 +1,11 @@
 package io.kestra.core.storages;
 
-import io.kestra.core.services.FlowService;
-import io.kestra.core.services.KVStoreService;
-import io.kestra.core.storages.kv.InternalKVStore;
-import io.kestra.core.storages.kv.KVStore;
+import io.kestra.core.services.NamespaceService;
 import jakarta.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.BufferedInputStream;
-import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
@@ -33,7 +29,8 @@ public class InternalStorage implements Storage {
     private final Logger logger;
     private final StorageContext context;
     private final StorageInterface storage;
-    private final FlowService flowService;
+    private final NamespaceFactory namespaceFactory;
+    private final NamespaceService namespaceService;
 
     /**
      * Creates a new {@link InternalStorage} instance.
@@ -41,8 +38,8 @@ public class InternalStorage implements Storage {
      * @param context The storage context.
      * @param storage The storage to delegate operations.
      */
-    public InternalStorage(StorageContext context, StorageInterface storage) {
-        this(LOG, context, storage, null);
+    public InternalStorage(StorageContext context, StorageInterface storage, NamespaceFactory namespaceFactory) {
+        this(LOG, context, storage, null, namespaceFactory);
     }
 
     /**
@@ -52,11 +49,12 @@ public class InternalStorage implements Storage {
      * @param context The storage context.
      * @param storage The storage to delegate operations.
      */
-    public InternalStorage(Logger logger, StorageContext context, StorageInterface storage, FlowService flowService) {
+    public InternalStorage(Logger logger, StorageContext context, StorageInterface storage, NamespaceService namespaceService, NamespaceFactory namespaceFactory) {
         this.logger = logger;
         this.context = context;
         this.storage = storage;
-        this.flowService = flowService;
+        this.namespaceService = namespaceService;
+        this.namespaceFactory = namespaceFactory;
     }
 
     /**
@@ -64,7 +62,7 @@ public class InternalStorage implements Storage {
      **/
     @Override
     public Namespace namespace() {
-        return new InternalNamespace(logger, context.getTenantId(), context.getNamespace(), storage);
+        return namespaceFactory.of(logger, context.getTenantId(), context.getNamespace(), storage);
     }
 
     /**
@@ -74,13 +72,13 @@ public class InternalStorage implements Storage {
     public Namespace namespace(String namespace) {
         boolean isExternalNamespace = !namespace.equals(context.getNamespace());
         // Checks whether the contextual namespace is allowed to access the passed namespace.
-        if (isExternalNamespace && flowService != null) {
-            flowService.checkAllowedNamespace(
+        if (isExternalNamespace && namespaceService != null) {
+            namespaceService.checkAllowedNamespace(
                 context.getTenantId(), namespace, // requested Tenant/Namespace
                 context.getTenantId(), context.getNamespace() // from Tenant/Namespace
             );
         }
-        return new InternalNamespace(logger, context.getTenantId(), namespace, storage);
+        return namespaceFactory.of(logger, context.getTenantId(), namespace, storage);
     }
 
     /**
@@ -102,6 +100,13 @@ public class InternalStorage implements Storage {
 
     }
 
+    @Override
+    public FileAttributes getAttributes(URI uri) throws IOException {
+        uriGuard(uri);
+
+        return this.storage.getAttributes(context.getTenantId(), context.getNamespace(), uri);
+    }
+
     /**
      * {@inheritDoc}
      **/
@@ -266,7 +271,13 @@ public class InternalStorage implements Storage {
         return this.storage.put(context.getTenantId(), context.getNamespace(), resolve, new BufferedInputStream(inputStream));
     }
 
+    @Override
     public Optional<StorageContext.Task> getTaskStorageContext() {
         return Optional.ofNullable((context instanceof StorageContext.Task task) ? task : null);
     }
+
+    @Override
+    public List<FileAttributes> list(URI uri) throws IOException {
+        return this.storage.list(context.getTenantId(), context.getNamespace(), uri);
+    }
 }

core/src/main/java/io/kestra/core/storages/Namespace.java

@@ -1,12 +1,22 @@
 package io.kestra.core.storages;
 
+import io.kestra.core.models.FetchVersion;
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
+import io.kestra.core.repositories.ArrayListTotal;
 import io.kestra.core.utils.PathMatcherPredicate;
+import io.micronaut.data.model.Pageable;
+import io.micronaut.data.model.Sort;
+import jakarta.annotation.Nullable;
+import org.apache.commons.lang3.tuple.Pair;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
+import java.time.ZonedDateTime;
+import java.util.Collections;
 import java.util.List;
 import java.util.function.Predicate;
 
@@ -16,6 +26,8 @@ import java.util.function.Predicate;
 public interface Namespace {
     String NAMESPACE_FILE_SCHEME = "nsfile";
 
+    ArrayListTotal<NamespaceFile> find(Pageable pageable, List<QueryFilter> filters, boolean allowDeleted, FetchVersion fetchVersion);
+
     /**
      * Gets the current namespace.
      *
@@ -37,19 +49,25 @@ public interface Namespace {
      */
     List<NamespaceFile> all() throws IOException;
 
-    /**
-     * Gets the URIs of all namespace files for the contextual namespace.
-     *
-     * @return The list of {@link URI}.
-     */
-    List<NamespaceFile> all(boolean includeDirectories) throws IOException;
+    default List<NamespaceFile> all(String containing) throws IOException {
+        return this.all(containing, false);
+    }
 
     /**
-     * Gets the URIs of all namespace files for the current namespace.
+     * Gets the URIs of all namespace files for the current namespace that contains the optional <code>containing</code> parameter.
      *
      * @return The list of {@link URI}.
      */
-    List<NamespaceFile> all(String prefix, boolean includeDirectories) throws IOException;
+    List<NamespaceFile> all(String containing, boolean includeDirectories) throws IOException;
+
+    /**
+     * Gets the URIs of all namespace files for the current namespace under the <code>parentPath</code>.
+     *
+     * @return The list of {@link URI}.
+     */
+    List<NamespaceFileMetadata> children(String parentPath, boolean recursive) throws IOException;
+
+    List<Pair<NamespaceFile, NamespaceFile>> move(Path source, Path target) throws Exception;
 
     /**
      * Gets a {@link NamespaceFile} for the given path and the current namespace.
@@ -57,7 +75,7 @@ public interface Namespace {
      * @param path the file path.
      * @return a new {@link NamespaceFile}
      */
-    NamespaceFile get(Path path);
+    NamespaceFile get(Path path) throws IOException;
 
     /**
      * Retrieves the URIs of all namespace files for the current namespace matching the given predicate.
@@ -82,27 +100,45 @@ public interface Namespace {
         return findAllFilesMatching(predicate);
     }
 
+    /**
+     * Retrieves the content of the namespace file at the given path for the latest version.
+     */
+    default InputStream getFileContent(Path path) throws IOException {
+        return getFileContent(path, null);
+    }
+
     /**
      * Retrieves the content of the namespace file at the given path.
      *
      * @param path the file path.
+     * @param version optionally a file version, otherwise will retrieve the latest.
      * @return the {@link InputStream}.
      * @throws IllegalArgumentException if the given {@link Path} is {@code null} or invalid.
      * @throws IOException              if an error happens while accessing the file.
      */
-    InputStream getFileContent(Path path) throws IOException;
+    InputStream getFileContent(Path path, @Nullable Integer version) throws IOException;
 
-    default NamespaceFile putFile(Path path, InputStream content) throws IOException, URISyntaxException {
+    /**
+     * Retrieves the metadata of the namespace file at the given path.
+     *
+     * @param path the file path.
+     * @return the {@link FileAttributes}.
+     */
+    FileAttributes getFileMetadata(Path path) throws IOException;
+
+    boolean exists(Path path) throws IOException;
+
+    default List<NamespaceFile> putFile(Path path, InputStream content) throws IOException, URISyntaxException {
         return putFile(path, content, Conflicts.OVERWRITE);
     }
 
-    NamespaceFile putFile(Path path, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException;
+    List<NamespaceFile> putFile(Path path, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException;
 
-    default NamespaceFile putFile(NamespaceFile file, InputStream content) throws IOException, URISyntaxException {
+    default List<NamespaceFile> putFile(NamespaceFile file, InputStream content) throws IOException, URISyntaxException {
         return putFile(file, content, Conflicts.OVERWRITE);
     }
 
-    default NamespaceFile putFile(NamespaceFile file, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException {
+    default List<NamespaceFile> putFile(NamespaceFile file, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException {
         return putFile(Path.of(file.path()), content, onAlreadyExist);
     }
 
@@ -110,39 +146,47 @@ public interface Namespace {
      * Creates a new directory for the current namespace.
      *
      * @param path The {@link Path} of the directory.
-     * @return The URI of the directory in the Kestra's internal storage.
+     * @return The created namespace file.
      * @throws IOException if an error happens while accessing the file.
      */
-    URI createDirectory(Path path) throws IOException;
+    NamespaceFile createDirectory(Path path) throws IOException;
 
     /**
-     * Deletes any namespaces files at the given path.
+     * Deletes any namespaces file at the given path.
      *
      * @param file the {@link NamespaceFile} to be deleted.
      * @throws IOException if an error happens while performing the delete operation.
      */
-    default boolean delete(NamespaceFile file) throws IOException {
+    default List<NamespaceFile> delete(NamespaceFile file) throws IOException {
         return delete(Path.of(file.path()));
     }
 
     /**
-     * Deletes namespaces directories at the given path.
-     *
-     * @param file the {@link NamespaceFile} to be deleted.
-     * @throws IOException if an error happens while performing the delete operation.
-     */
-    default boolean deleteDirectory(NamespaceFile file) throws IOException {
-        return delete(Path.of(file.path()));
-    }
-
-    /**
-     * Deletes any namespaces files at the given path.
+     * Soft-deletes any namespaces files at the given path.
      *
      * @param path the path to be deleted.
-     * @return {@code true} if the file was deleted by this method; {@code false} if the file could not be deleted because it did not exist
+     * @return the list of namespace files that got deleted. There can be multiple files if a directory is deleted as its whole content will be.
      * @throws IOException if an error happens while performing the delete operation.
      */
-    boolean delete(Path path) throws IOException;
+    List<NamespaceFile> delete(Path path) throws IOException;
+
+    /**
+     * Hard-deletes any namespaces files.
+     *
+     * @param namespaceFile the namespace file to be purged.
+     * @return {@code true} if the file was purged by this method; {@code false} if the file could not be deleted because it did not exist
+     * @throws IOException if an error happens while performing the delete operation.
+     */
+    boolean purge(NamespaceFile namespaceFile) throws IOException;
+
+    /**
+     * Hard-deletes all provided namespaces files.
+     *
+     * @param namespaceFiles the namespace files to be purged.
+     * @return the amount of files that were purged.
+     * @throws IOException if an error happens while performing the delete operation.
+     */
+    Integer purge(List<NamespaceFile> namespaceFiles) throws IOException;
 
     /**
      * Checks if a directory is empty.

core/src/main/java/io/kestra/core/storages/NamespaceFactory.java

@@ -0,0 +1,20 @@
+package io.kestra.core.storages;
+
+import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.slf4j.Logger;
+
+@Singleton
+public class NamespaceFactory {
+    @Inject
+    private NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepositoryInterface;
+
+    public Namespace of(String tenantId, String namespace, StorageInterface storageInterface) {
+        return new InternalNamespace(tenantId, namespace, storageInterface, namespaceFileMetadataRepositoryInterface);
+    }
+
+    public Namespace of(Logger logger, String tenantId, String namespace, StorageInterface storageInterface) {
+        return new InternalNamespace(logger, tenantId, namespace, storageInterface, namespaceFileMetadataRepositoryInterface);
+    }
+}

core/src/main/java/io/kestra/core/storages/NamespaceFile.java

@@ -1,11 +1,14 @@
 package io.kestra.core.storages;
 
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
 import io.kestra.core.utils.WindowsUtils;
 import jakarta.annotation.Nullable;
 
 import java.net.URI;
 import java.nio.file.Path;
 import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * Represents a NamespaceFile object.
@@ -13,15 +16,22 @@ import java.util.Objects;
  * @param path      The path of file relative to the namespace.
  * @param uri       The URI of the namespace file in the Kestra's internal storage.
  * @param namespace The namespace of the file.
+ * @param version The version of the file.
  */
 public record NamespaceFile(
     String path,
     URI uri,
-    String namespace
+    String namespace,
+    int version
 ) {
+    private static final Pattern capturePathWithoutVersion = Pattern.compile("(.*)(?:\\.v\\d+)?$");
 
     public NamespaceFile(Path path, URI uri, String namespace) {
-        this(path.toString(), uri, namespace);
+        this(path.toString(), uri, namespace, 1);
+    }
+
+    public NamespaceFile(String path, URI uri, String namespace) {
+        this(path, uri, namespace, 1);
     }
 
     /**
@@ -33,7 +43,19 @@ public record NamespaceFile(
      * @return a new {@link NamespaceFile} object
      */
     public static NamespaceFile of(final String namespace) {
-        return of(namespace, (Path) null);
+        return of(namespace, (Path) null, 1);
+    }
+
+    public static NamespaceFile of(final String namespace, final URI uri) {
+        return of(namespace, uri, 1);
+    }
+
+    public static NamespaceFile fromMetadata(final NamespaceFileMetadata metadata) {
+        return of(
+            metadata.getNamespace(),
+            Path.of(metadata.getPath()),
+            metadata.getVersion()
+        );
     }
 
     /**
@@ -43,9 +65,9 @@ public record NamespaceFile(
      * @param namespace The namespace - cannot be {@code null}.
      * @return a new {@link NamespaceFile} object
      */
-    public static NamespaceFile of(final String namespace, @Nullable final URI uri) {
+    public static NamespaceFile of(final String namespace, @Nullable final URI uri, int version) {
         if (uri == null || uri.equals(URI.create("/"))) {
-            return of(namespace, (Path) null);
+            return of(namespace, (Path) null, version);
         }
 
         Path path = Path.of(WindowsUtils.windowsToUnixPath(uri.getPath()));
@@ -61,9 +83,9 @@ public record NamespaceFile(
                     "Invalid Kestra URI. Expected prefix for namespace '%s', but was %s.", namespace, uri)
                 );
             }
-            namespaceFile = of(namespace, Path.of(StorageContext.namespaceFilePrefix(namespace)).relativize(path));
+            namespaceFile = of(namespace, Path.of(StorageContext.namespaceFilePrefix(namespace)).relativize(path), version);
         } else {
-            namespaceFile = of(namespace, path);
+            namespaceFile = of(namespace, path, version);
         }
 
         boolean trailingSlash = uri.toString().endsWith("/");
@@ -75,10 +97,15 @@ public record NamespaceFile(
         return new NamespaceFile(
             namespaceFile.path,
             URI.create(namespaceFile.uri.toString() + "/"),
-            namespaceFile.namespace
+            namespaceFile.namespace,
+            version
         );
     }
 
+    public static NamespaceFile of(final String namespace, final Path path) {
+        return of(namespace, path, 1);
+    }
+
     /**
      * Static factory method for constructing a new {@link NamespaceFile} object.
      *
@@ -86,31 +113,61 @@ public record NamespaceFile(
      * @param namespace The namespace - cannot be {@code null}.
      * @return a new {@link NamespaceFile} object
      */
-    public static NamespaceFile of(final String namespace, @Nullable final Path path) {
+    public static NamespaceFile of(final String namespace, @Nullable final Path path, int version) {
         Objects.requireNonNull(namespace, "namespace cannot be null");
         if (path == null || path.equals(Path.of("/"))) {
             return new NamespaceFile(
                 "",
                 URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.namespaceFilePrefix(namespace) + "/"),
-                namespace
+                namespace,
+                // Directory always has a single version
+                1
             );
         }
 
+        return of(namespace, path.toString(), version);
+    }
+
+    public static NamespaceFile of(String namespace, String path, int version) {
         Path namespacePrefixPath = Path.of(StorageContext.namespaceFilePrefix(namespace));
-        Path filePath = path.normalize();
-        if (filePath.isAbsolute()) {
-            filePath = filePath.getRoot().relativize(filePath);
-        }
         // Need to remove starting trailing slash for Windows
-        String pathWithoutTrailingSlash = path.toString().replaceFirst("^[.]*[\\\\|/]+", "");
+        String pathWithoutLeadingSlash = path.replaceFirst("^[.]*[\\\\|/]+", "");
+
+        version = NamespaceFile.isDirectory(pathWithoutLeadingSlash) ? 1 : version;
+
+        String storagePath = pathWithoutLeadingSlash;
+        if (!pathWithoutLeadingSlash.endsWith("/") && version > 1) {
+            storagePath += ".v" + version;
+        }
 
         return new NamespaceFile(
-            pathWithoutTrailingSlash,
-            URI.create(StorageContext.KESTRA_PROTOCOL + namespacePrefixPath.resolve(pathWithoutTrailingSlash).toString().replace("\\","/")),
-            namespace
+            pathWithoutLeadingSlash,
+            URI.create(StorageContext.KESTRA_PROTOCOL + namespacePrefixPath.resolve(storagePath).toString().replace("\\", "/")),
+            namespace,
+            version
         );
     }
 
+    public static Path normalize(String pathStr, boolean withLeadingSlash) {
+        return normalize(Path.of(pathStr), withLeadingSlash);
+    }
+
+    public static Path normalize(Path path, boolean withLeadingSlash) {
+        if (path == null) {
+            return Path.of("/");
+        }
+
+        if (withLeadingSlash && !path.toString().startsWith("/")) {
+            return Path.of("/" + path);
+        }
+
+        if (!withLeadingSlash && path.toString().startsWith("/")) {
+            return Path.of(path.toString().substring(1));
+        }
+
+        return path;
+    }
+
     /**
      * Returns the path of file relative to the namespace.
      *
@@ -118,17 +175,13 @@ public record NamespaceFile(
      * @return The path.
      */
     public Path path(boolean withLeadingSlash) {
-        final String strPath = path.toString();
-        if (!withLeadingSlash) {
-            if (strPath.startsWith("/")) {
-                return Path.of(strPath.substring(1));
-            }
-        } else {
-            if (!strPath.startsWith("/")) {
-                return Path.of("/").resolve(path);
-            }
+        String strPath = path;
+        Matcher matcher = capturePathWithoutVersion.matcher(strPath);
+        if (matcher.matches()) {
+            strPath = matcher.group(1);
         }
-        return Path.of(path);
+
+        return normalize(Path.of(strPath), withLeadingSlash);
     }
 
     /**
@@ -147,8 +200,12 @@ public record NamespaceFile(
      *
      * @return {@code true} if this namespace file is a directory.
      */
+    public static boolean isDirectory(String path) {
+        return path.endsWith("/");
+    }
+
     public boolean isDirectory() {
-        return uri.toString().endsWith("/");
+        return isDirectory(uri.toString());
     }
 
     /**

core/src/main/java/io/kestra/core/storages/NamespaceFileAttributes.java

@@ -0,0 +1,54 @@
+package io.kestra.core.storages;
+
+import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+
+public class NamespaceFileAttributes implements FileAttributes {
+    private final NamespaceFileMetadata namespaceFileMetadata;
+
+    public NamespaceFileAttributes(NamespaceFileMetadata namespaceFileMetadata) {
+        this.namespaceFileMetadata = namespaceFileMetadata;
+    }
+
+    @Override
+    public String getFileName() {
+        String name = new File(namespaceFileMetadata.getPath()).getName();
+
+        if (name.isEmpty()) {
+            return "_files";
+        }
+
+        return name;
+    }
+
+    @Override
+    public long getLastModifiedTime() {
+        return Optional.ofNullable(namespaceFileMetadata.getUpdated()).map(Instant::toEpochMilli).orElse(0L);
+    }
+
+    @Override
+    public long getCreationTime() {
+        return Optional.ofNullable(namespaceFileMetadata.getCreated()).map(Instant::toEpochMilli).orElse(0L);
+    }
+
+    @Override
+    public FileType getType() {
+        return namespaceFileMetadata.getPath().endsWith("/") ? FileType.Directory : FileType.File;
+    }
+
+    @Override
+    public long getSize() {
+        return namespaceFileMetadata.getSize();
+    }
+
+    @Override
+    public Map<String, String> getMetadata() throws IOException {
+        return Collections.emptyMap();
+    }
+}

core/src/main/java/io/kestra/core/storages/NamespaceFileRevision.java

@@ -0,0 +1,3 @@
+package io.kestra.core.storages;
+
+public record NamespaceFileRevision(Integer revision) {}

core/src/main/java/io/kestra/core/storages/Storage.java

@@ -1,8 +1,10 @@
 package io.kestra.core.storages;
 
+import io.kestra.core.annotations.Retryable;
 import jakarta.annotation.Nullable;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
@@ -46,6 +48,15 @@ public interface Storage {
      */
     InputStream getFile(URI uri) throws IOException;
 
+    /**
+     * Retrieves the metadata attributes for the given URI.
+     *
+     * @param uri       the URI of the object
+     * @return the file attributes
+     * @throws IOException if the attributes cannot be retrieved
+     */
+    FileAttributes getAttributes(URI uri) throws IOException;
+
     /**
      * Deletes the file for the given URI.
      * @param uri the file URI.
@@ -162,4 +173,6 @@ public interface Storage {
      * @return the task storage context
      */
     Optional<StorageContext.Task> getTaskStorageContext();
+
+    List<FileAttributes> list(URI uri) throws IOException;
 }

core/src/main/java/io/kestra/core/storages/StorageInterface.java

@@ -13,6 +13,7 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
+import java.nio.file.NoSuchFileException;
 import java.util.List;
 
 /**
@@ -52,7 +53,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return an InputStream to read the object's contents
      * @throws IOException if the object cannot be read
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     InputStream get(String tenantId, @Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -64,7 +65,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return an InputStream to read the object's contents
      * @throws IOException if the object cannot be read
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     InputStream getInstanceResource(@Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -76,7 +77,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return the storage object with metadata
      * @throws IOException if the object cannot be retrieved
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     StorageObject getWithMetadata(String tenantId, @Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -89,7 +90,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return a list of matching object URIs
      * @throws IOException if the listing fails
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     List<URI> allByPrefix(String tenantId, @Nullable String namespace, URI prefix, boolean includeDirectories) throws IOException;
 
     /**
@@ -101,7 +102,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return a list of file attributes
      * @throws IOException if the listing fails
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     List<FileAttributes> list(String tenantId, @Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -113,7 +114,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return a list of file attributes
      * @throws IOException if the listing fails
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     List<FileAttributes> listInstanceResource(@Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -159,7 +160,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return the file attributes
      * @throws IOException if the attributes cannot be retrieved
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     FileAttributes getAttributes(String tenantId, @Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -171,7 +172,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return the file attributes
      * @throws IOException if the attributes cannot be retrieved
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     FileAttributes getInstanceAttributes(@Nullable String namespace, URI uri) throws IOException;
 
     /**
@@ -288,7 +289,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
      * @return the URI of the moved object
      * @throws IOException if moving fails
      */
-    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
+    @Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
     URI move(String tenantId, @Nullable String namespace, URI from, URI to) throws IOException;
 
     /**

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

@@ -6,6 +6,10 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BooleanSupplier;
 import java.util.function.Supplier;
 
+/**
+ * @deprecated use {@link org.awaitility.Awaitility} instead
+ */
+@Deprecated
 public class Await {
     private static final Duration defaultSleep = Duration.ofMillis(100);
 

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

@@ -1,13 +1,39 @@
 package io.kestra.core.utils;
 
+import io.kestra.core.models.Setting;
+import io.kestra.core.repositories.SettingRepositoryInterface;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 
+import java.util.Optional;
+
 @Singleton
 public class EditionProvider {
     public Edition get() {
         return Edition.OSS;
     }
 
+    @Inject
+    private Optional<SettingRepositoryInterface> settingRepository; // repositories are not always there on unit tests
+
+    @PostConstruct
+    void start() {
+        // check the edition in the settings and update if needed, we didn't use it would allow us to detect incompatible update later if needed
+        settingRepository.ifPresent(settingRepositoryInterface -> persistEdition(settingRepositoryInterface, get()));
+    }
+
+    private void persistEdition(SettingRepositoryInterface settingRepositoryInterface, Edition edition) {
+        Optional<Setting> versionSetting = settingRepositoryInterface.findByKey(Setting.INSTANCE_EDITION);
+        if (versionSetting.isEmpty() || !versionSetting.get().getValue().equals(edition)) {
+            settingRepositoryInterface.save(Setting.builder()
+                .key(Setting.INSTANCE_EDITION)
+                .value(edition)
+                .build()
+            );
+        }
+    }
+
     public enum Edition {
         OSS,
         EE

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

@@ -11,11 +11,14 @@ import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 
+/**
+ * Utility class to create {@link java.util.concurrent.ExecutorService} with {@link java.util.concurrent.ExecutorService} instances.
+ * WARNING: those instances will use the {@link ThreadUncaughtExceptionHandler} which terminates Kestra if an error occurs in any thread,
+ * so it should not be used inside plugins.
+ */
 @Singleton
 @Slf4j
 public class ExecutorsUtils {
-    @Inject
-    private ThreadMainFactoryBuilder threadFactoryBuilder;
 
     @Inject
     private MeterRegistry meterRegistry;
@@ -24,7 +27,7 @@ public class ExecutorsUtils {
         return this.wrap(
             name,
             Executors.newCachedThreadPool(
-                threadFactoryBuilder.build(name + "_%d")
+                ThreadMainFactoryBuilder.build(name + "_%d")
             )
         );
     }
@@ -36,7 +39,7 @@ public class ExecutorsUtils {
             60L,
             TimeUnit.SECONDS,
             new LinkedBlockingQueue<>(),
-            threadFactoryBuilder.build(name + "_%d")
+            ThreadMainFactoryBuilder.build(name + "_%d")
         );
 
         threadPoolExecutor.allowCoreThreadTimeOut(true);
@@ -51,7 +54,7 @@ public class ExecutorsUtils {
         return this.wrap(
             name,
             Executors.newSingleThreadExecutor(
-                threadFactoryBuilder.build(name + "_%d")
+                ThreadMainFactoryBuilder.build(name + "_%d")
             )
         );
     }
@@ -60,7 +63,7 @@ public class ExecutorsUtils {
         return this.wrap(
             name,
             Executors.newSingleThreadScheduledExecutor(
-                threadFactoryBuilder.build(name + "_%d")
+                ThreadMainFactoryBuilder.build(name + "_%d")
             )
         );
     }

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

@@ -65,10 +65,9 @@ public class ListUtils {
     }
 
     public static List<String> convertToListString(Object object){
-        if (object instanceof List<?> list && (list.isEmpty() || list.getFirst() instanceof String)) {
-            return (List<String>) list;
-        } else {
-            throw new IllegalArgumentException("%s in not an instance of List of String".formatted(object));
-        }
+        return convertToList(object)
+            .stream()
+            .map(Object::toString)
+            .toList();
     }
 }

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

@@ -1,38 +1,27 @@
-package io.kestra.core.services;
+package io.kestra.core.utils;
 
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.FlowId;
 import io.kestra.core.models.triggers.TriggerContext;
-import io.kestra.core.repositories.LogRepositoryInterface;
-import io.micronaut.data.model.Pageable;
-import io.micronaut.data.model.Sort;
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
 import org.apache.commons.lang3.ArrayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.event.Level;
 
-import java.time.ZonedDateTime;
-import java.util.List;
+/**
+ * Utility class for server logging
+ */
+public final class Logs {
 
-@Singleton
-public class LogService {
     private static final String FLOW_PREFIX_WITH_TENANT = "[tenant: {}] [namespace: {}] [flow: {}] ";
     private static final String EXECUTION_PREFIX_WITH_TENANT = FLOW_PREFIX_WITH_TENANT + "[execution: {}] ";
     private static final String TRIGGER_PREFIX_WITH_TENANT = FLOW_PREFIX_WITH_TENANT + "[trigger: {}] ";
     private static final String TASKRUN_PREFIX_WITH_TENANT = FLOW_PREFIX_WITH_TENANT + "[task: {}] [execution: {}] [taskrun: {}] ";
 
-    private final LogRepositoryInterface logRepository;
+    private Logs() {}
 
-    @Inject
-    public LogService(LogRepositoryInterface logRepository) {
-        this.logRepository = logRepository;
-    }
-
-    public void logExecution(FlowId flow, Logger logger, Level level, String message, Object... args) {
+    public static void logExecution(FlowId flow, Logger logger, Level level, String message, Object... args) {
         String finalMsg = FLOW_PREFIX_WITH_TENANT + message;
         Object[] executionArgs = new Object[] { flow.getTenantId(), flow.getNamespace(), flow.getId() };
         Object[] finalArgs = ArrayUtils.addAll(executionArgs, args);
@@ -40,37 +29,37 @@ public class LogService {
     }
 
     /**
-     * Log an execution via the execution logger named: 'execution.{flowId}'.
+     * Log an {@link Execution} via the executor logger named: 'executor.{tenantId}.{namespace}.{flowId}'.
      */
-    public void logExecution(Execution execution, Level level, String message, Object... args) {
+    public static void logExecution(Execution execution, Level level, String message, Object... args) {
         Logger logger = logger(execution);
         logExecution(execution, logger, level, message, args);
     }
 
-    public void logExecution(Execution execution, Logger logger, Level level, String message, Object... args) {
+    public static void logExecution(Execution execution, Logger logger, Level level, String message, Object... args) {
         Object[] executionArgs = new Object[] { execution.getTenantId(), execution.getNamespace(), execution.getFlowId(), execution.getId() };
         Object[] finalArgs = ArrayUtils.addAll(executionArgs, args);
         logger.atLevel(level).log(EXECUTION_PREFIX_WITH_TENANT + message, finalArgs);
     }
 
     /**
-     * Log a trigger via the trigger logger named: 'trigger.{flowId}.{triggereId}'.
+     * Log a {@link TriggerContext} via the scheduler logger named: 'trigger.{tenantId}.{namespace}.{flowId}.{triggerId}'.
      */
-    public void logTrigger(TriggerContext triggerContext, Level level, String message, Object... args) {
+    public static void logTrigger(TriggerContext triggerContext, Level level, String message, Object... args) {
         Logger logger = logger(triggerContext);
         logTrigger(triggerContext, logger, level, message, args);
     }
 
-    public void logTrigger(TriggerContext triggerContext, Logger logger, Level level, String message, Object... args) {
+    public static void logTrigger(TriggerContext triggerContext, Logger logger, Level level, String message, Object... args) {
         Object[] executionArgs = new Object[] { triggerContext.getTenantId(), triggerContext.getNamespace(), triggerContext.getFlowId(), triggerContext.getTriggerId() };
         Object[] finalArgs = ArrayUtils.addAll(executionArgs, args);
         logger.atLevel(level).log(TRIGGER_PREFIX_WITH_TENANT + message, finalArgs);
     }
 
     /**
-     * Log a taskRun via the taskRun logger named: 'task.{flowId}.{taskId}'.
+     * Log a {@link TaskRun} via the worker logger named: 'worker.{tenantId}.{namespace}.{flowId}.{taskId}'.
      */
-    public void logTaskRun(TaskRun taskRun, Level level, String message, Object... args) {
+    public static void logTaskRun(TaskRun taskRun, Level level, String message, Object... args) {
         String prefix = TASKRUN_PREFIX_WITH_TENANT;
         String finalMsg = taskRun.getValue() == null ? prefix + message : prefix + "[value: {}] " + message;
         Object[] executionArgs = new Object[] { taskRun.getTenantId(), taskRun.getNamespace(), taskRun.getFlowId(), taskRun.getTaskId(), taskRun.getExecutionId(), taskRun.getId() };
@@ -82,33 +71,21 @@ public class LogService {
         logger.atLevel(level).log(finalMsg, finalArgs);
     }
 
-    public int purge(String tenantId, String namespace, String flowId, String executionId, List<Level> logLevels, ZonedDateTime startDate, ZonedDateTime endDate) {
-        return logRepository.deleteByQuery(tenantId, namespace, flowId, executionId, logLevels, startDate, endDate);
-    }
-
-    /**
-     * Fetch the error logs of an execution.
-     * Will limit the results to the first 25 error logs, ordered by timestamp asc.
-     */
-    public List<LogEntry> errorLogs(String tenantId, String executionId) {
-        return logRepository.findByExecutionId(tenantId, executionId, Level.ERROR, Pageable.from(1, 25, Sort.of(Sort.Order.asc("timestamp"))));
-    }
-
-    private Logger logger(TaskRun taskRun) {
+    private static Logger logger(TaskRun taskRun) {
         return LoggerFactory.getLogger(
-            "task." + taskRun.getFlowId() + "." + taskRun.getTaskId()
+            "worker." + taskRun.getTenantId() + "." + taskRun.getNamespace() + "." + taskRun.getFlowId() + "." + taskRun.getTaskId()
         );
     }
 
-    private Logger logger(TriggerContext triggerContext) {
+    private static Logger logger(TriggerContext triggerContext) {
         return LoggerFactory.getLogger(
-            "trigger." + triggerContext.getFlowId() + "." + triggerContext.getTriggerId()
+            "scheduler." + triggerContext.getTenantId() + "." + triggerContext.getNamespace() + "." + triggerContext.getFlowId() + "." + triggerContext.getTriggerId()
         );
     }
 
-    private Logger logger(Execution execution) {
+    private static Logger logger(Execution execution) {
         return LoggerFactory.getLogger(
-            "execution." + execution.getFlowId()
+            "executor." + execution.getTenantId() + "." + execution.getNamespace() + "." + execution.getFlowId()
         );
     }
 }

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

@@ -120,7 +120,10 @@ public class MapUtils {
     private static Collection<?> mergeCollections(Collection<?> colA, Collection<?> colB) {
         List<Object> merged = new ArrayList<>(colA.size() + colB.size());
         merged.addAll(colA);
-        merged.addAll(colB);
+        if (!colB.isEmpty()) {
+            List<?> filtered = colB.stream().filter(it -> !colA.contains(it)).toList();
+            merged.addAll(filtered);
+        }
         return merged;
     }
 

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

@@ -1,14 +1,12 @@
 package io.kestra.core.utils;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.kestra.core.models.tasks.FileExistComportment;
 import io.kestra.core.models.tasks.NamespaceFiles;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.storages.NamespaceFile;
-import jakarta.annotation.PostConstruct;
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.time.DurationFormatUtils;
 import org.apache.commons.lang3.time.StopWatch;
@@ -19,26 +17,27 @@ import java.io.InputStream;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.*;
 
 import static io.kestra.core.utils.Rethrow.throwConsumer;
 
-@Singleton
-public class NamespaceFilesUtils {
-    @Inject
-    private ExecutorsUtils executorsUtils;
+public final class NamespaceFilesUtils {
+    private static final int maxThreads = Math.max(Runtime.getRuntime().availableProcessors() * 4, 32);
+    private static final ExecutorService EXECUTOR_SERVICE = new ThreadPoolExecutor(
+        0,
+        maxThreads,
+        60L,
+        TimeUnit.SECONDS,
+        new SynchronousQueue<>(),
+        new ThreadFactoryBuilder().setNameFormat("namespace-files").build()
+    );;
 
-    private ExecutorService executorService;
-
-    @PostConstruct
-    public void postConstruct() {
-        this.executorService = executorsUtils.maxCachedThreadPool(Math.max(Runtime.getRuntime().availableProcessors() * 4, 32), "namespace-file");
+    private NamespaceFilesUtils() {
+        // utility class pattern
     }
 
-    public void loadNamespaceFiles(
+    public static void loadNamespaceFiles(
         RunContext runContext,
         NamespaceFiles namespaceFiles
     )
@@ -63,7 +62,11 @@ public class NamespaceFilesUtils {
           matchedNamespaceFiles.addAll(files);
         }
 
+        // Use half of the available threads to avoid impacting concurrent tasks
+        int parallelism = maxThreads / 2;
         Flux.fromIterable(matchedNamespaceFiles)
+            .parallel(parallelism)
+            .runOn(Schedulers.fromExecutorService(EXECUTOR_SERVICE))
             .doOnNext(throwConsumer(nsFile -> {
                 InputStream content = runContext.storage().getFile(nsFile.uri());
                 Path path = folderPerNamespace ?
@@ -71,7 +74,7 @@ public class NamespaceFilesUtils {
                     Path.of(nsFile.path());
                 runContext.workingDir().putFile(path, content, fileExistComportment);
             }))
-            .publishOn(Schedulers.fromExecutorService(executorService))
+            .sequential()
             .blockLast();
 
         Duration duration = stopWatch.getDuration();

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

@@ -17,36 +17,36 @@ import org.slf4j.Logger;
 
 import java.io.Serial;
 import java.time.Duration;
-import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.function.BiPredicate;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
-import jakarta.inject.Singleton;
+public final class RetryUtils {
+    private RetryUtils() {
+        // utility class pattern
+    }
 
-@Singleton
-public class RetryUtils {
-    public <T, E extends Throwable> Instance<T, E> of() {
+    public static <T, E extends Throwable> Instance<T, E> of() {
         return Instance.<T, E>builder()
             .build();
     }
 
-    public <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy) {
+    public static <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy) {
         return Instance.<T, E>builder()
             .policy(policy)
             .build();
     }
 
-    public <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Function<RetryFailed, E> failureFunction) {
+    public static <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Function<RetryFailed, E> failureFunction) {
         return Instance.<T, E>builder()
             .policy(policy)
             .failureFunction(failureFunction)
             .build();
     }
 
-    public <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Logger logger) {
+    public static <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Logger logger) {
         return Instance.<T, E>builder()
             .policy(policy)
             .logger(logger)
@@ -199,7 +199,6 @@ public class RetryUtils {
 
         private final int attemptCount;
         private final Duration elapsedTime;
-        private final Instant startTime;
 
         public <T> RetryFailed(ExecutionAttemptedEvent<? extends T> event) {
             super(
@@ -210,7 +209,6 @@ public class RetryUtils {
 
             this.attemptCount = event.getAttemptCount();
             this.elapsedTime = event.getElapsedTime();
-            this.startTime = event.getStartTime().get();
         }
     }
 }