fake commit to run flaky tests

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
Co-authored-by: Bart Ledoux <bledoux@kestra.io>

.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,73 @@ 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/*"]
+
       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/*",
+            "vitest",
+            "@vitest/*",
+            # Temporary exclusion of these packages from major updates
+            "eslint-plugin-storybook",
+            "eslint-plugin-vue",
+          ]
+
+      minor:
+        update-types: ["minor"]
+        applies-to: version-updates
+        exclude-patterns: [
+            "@esbuild/*",
+            "@rollup/*",
+            "@swc/*",
+            "@types/*",
+            "storybook*",
+            "@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",
-          ]
-        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/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.gradle

@@ -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.1.0.6387"
     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 ->

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

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

@@ -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())
@@ -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/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

@@ -11,6 +11,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;
@@ -129,6 +130,14 @@ public class Flow extends AbstractFlow implements HasUID {
     @Valid
     @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(

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

@@ -43,6 +43,7 @@ public class FlowWithSource extends Flow {
             .concurrency(this.concurrency)
             .retry(this.retry)
             .sla(this.sla)
+            .checks(this.checks)
             .build();
     }
 
@@ -85,6 +86,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();
         }

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,9 +89,6 @@ 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
      */
@@ -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/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/TriggerRepositoryInterface.java

@@ -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

@@ -123,7 +123,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;
     }
@@ -574,6 +579,11 @@ public class DefaultRunContext extends RunContext {
         return isInitialized.get();
     }
 
+    @Override
+    public AclChecker acl() {
+        return new AclCheckerImpl(this.applicationContext, flowInfo());
+    }
+
     @Override
     public LocalPath localPath() {
         return localPath;

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();
     }
 

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

@@ -192,5 +192,16 @@ 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();
 }

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

@@ -8,8 +8,9 @@ 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,7 +45,10 @@ 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;
@@ -135,7 +139,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 +234,8 @@ public class RunContextInitializer {
             runContextLogger.logger(),
             context,
             storageInterface,
-            flowService
+            namespaceService,
+            namespaceFactory
         );
 
         runContext.setLogger(runContextLogger);

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/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>
@@ -85,6 +93,50 @@ public class FlowService {
             .orElseThrow(() -> new IllegalStateException("Cannot perform operation on flow. Cause: No FlowRepository"));
     }
 
+    /**
+     * 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.
      * <p>
@@ -456,50 +508,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

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,12 @@
 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.repositories.NamespaceFileMetadataRepositoryInterface;
+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 +30,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 +39,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 +50,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 +63,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 +73,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 +101,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}
      **/

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.

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

@@ -14,8 +14,6 @@ import lombok.extern.slf4j.Slf4j;
 @Singleton
 @Slf4j
 public class ExecutorsUtils {
-    @Inject
-    private ThreadMainFactoryBuilder threadFactoryBuilder;
 
     @Inject
     private MeterRegistry meterRegistry;
@@ -24,7 +22,7 @@ public class ExecutorsUtils {
         return this.wrap(
             name,
             Executors.newCachedThreadPool(
-                threadFactoryBuilder.build(name + "_%d")
+                ThreadMainFactoryBuilder.build(name + "_%d")
             )
         );
     }
@@ -36,7 +34,7 @@ public class ExecutorsUtils {
             60L,
             TimeUnit.SECONDS,
             new LinkedBlockingQueue<>(),
-            threadFactoryBuilder.build(name + "_%d")
+            ThreadMainFactoryBuilder.build(name + "_%d")
         );
 
         threadPoolExecutor.allowCoreThreadTimeOut(true);
@@ -51,7 +49,7 @@ public class ExecutorsUtils {
         return this.wrap(
             name,
             Executors.newSingleThreadExecutor(
-                threadFactoryBuilder.build(name + "_%d")
+                ThreadMainFactoryBuilder.build(name + "_%d")
             )
         );
     }
@@ -60,7 +58,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/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 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 Logs() {}
 
-    private final LogRepositoryInterface logRepository;
-
-    @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 execution logger named: 'execution.{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 trigger logger named: 'trigger.{flowId}.{triggereId}'.
      */
-    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 taskRun logger named: 'task.{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,31 +71,19 @@ 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()
         );
     }
 
-    private Logger logger(TriggerContext triggerContext) {
+    private static Logger logger(TriggerContext triggerContext) {
         return LoggerFactory.getLogger(
             "trigger." + triggerContext.getFlowId() + "." + triggerContext.getTriggerId()
         );
     }
 
-    private Logger logger(Execution execution) {
+    private static Logger logger(Execution execution) {
         return LoggerFactory.getLogger(
             "execution." + execution.getFlowId()
         );

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();
         }
     }
 }

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

@@ -3,18 +3,18 @@ package io.kestra.core.utils;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import java.util.concurrent.ThreadFactory;
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
 
-@Singleton
-public class ThreadMainFactoryBuilder {
-    @Inject
-    private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
 
-    public ThreadFactory build(String name) {
+public final class ThreadMainFactoryBuilder {
+
+    private ThreadMainFactoryBuilder() {
+        // utility class pattern
+    }
+
+    public static ThreadFactory build(String name) {
         return new ThreadFactoryBuilder()
             .setNameFormat(name)
-            .setUncaughtExceptionHandler(this.uncaughtExceptionHandler)
+            .setUncaughtExceptionHandler(ThreadUncaughtExceptionHandler.INSTANCE)
             .build();
     }
 }

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

@@ -1,27 +1,21 @@
 package io.kestra.core.utils;
 
-import io.micronaut.context.ApplicationContext;
+import io.kestra.core.contexts.KestraContext;
 import lombok.extern.slf4j.Slf4j;
 
 import java.lang.Thread.UncaughtExceptionHandler;
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
 
 @Slf4j
-@Singleton
-public final class ThreadUncaughtExceptionHandlers implements UncaughtExceptionHandler {
-    @Inject
-    private ApplicationContext applicationContext;
-
-    private final Runtime runtime = Runtime.getRuntime();
+public final class ThreadUncaughtExceptionHandler implements UncaughtExceptionHandler {
+    public static final UncaughtExceptionHandler INSTANCE = new ThreadUncaughtExceptionHandler();
 
     @Override
     public void uncaughtException(Thread t, Throwable e) {
-        boolean isTest = applicationContext.getEnvironment().getActiveNames().contains("test");
+        boolean isTest = KestraContext.getContext().getEnvironments().contains("test");
 
         try {
             // cannot use FormattingLogger due to a dependency loop
-            log.error("Caught an exception in {}. " + (isTest ? "Keeping it running for test." : "Shutting down."), t, e);
+            log.error("Caught an exception in {}. {}", t, isTest ? "Keeping it running for test." : "Shutting down.", e);
         } catch (Throwable errorInLogging) {
             // If logging fails, e.g. due to missing memory, at least try to log the
             // message and the cause for the failed logging.
@@ -29,8 +23,8 @@ public final class ThreadUncaughtExceptionHandlers implements UncaughtExceptionH
             System.err.println(errorInLogging.getMessage());
         } finally {
             if (!isTest) {
-                applicationContext.close();
-                runtime.exit(1);
+                KestraContext.getContext().shutdown();
+                Runtime.getRuntime().exit(1);
             }
         }
     }

core/src/main/java/io/kestra/core/validations/FilesVersionBehaviorValidation.java

@@ -0,0 +1,16 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.validations.validator.FilesVersionBehaviorValidator;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = FilesVersionBehaviorValidator.class)
+public @interface FilesVersionBehaviorValidation {
+    String message() default "invalid `version` behavior configuration";
+    Class<?>[] groups() default {};
+    Class<? extends Payload>[] payload() default {};
+}

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

@@ -0,0 +1,35 @@
+package io.kestra.core.validations.validator;
+
+import io.kestra.core.validations.FilesVersionBehaviorValidation;
+import io.kestra.core.validations.KvVersionBehaviorValidation;
+import io.kestra.plugin.core.namespace.Version;
+import io.micronaut.core.annotation.AnnotationValue;
+import io.micronaut.core.annotation.Introspected;
+import io.micronaut.core.annotation.NonNull;
+import io.micronaut.core.annotation.Nullable;
+import io.micronaut.validation.validator.constraints.ConstraintValidator;
+import io.micronaut.validation.validator.constraints.ConstraintValidatorContext;
+import jakarta.inject.Singleton;
+
+@Singleton
+@Introspected
+public class FilesVersionBehaviorValidator implements ConstraintValidator<FilesVersionBehaviorValidation, Version> {
+    @Override
+    public boolean isValid(
+        @Nullable Version value,
+        @NonNull AnnotationValue<FilesVersionBehaviorValidation> annotationMetadata,
+        @NonNull ConstraintValidatorContext context) {
+        if (value == null) {
+            return true;
+        }
+
+        if (value.getBefore() != null && value.getKeepAmount() != null) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate("Cannot set both 'before' and 'keepAmount' properties")
+                .addConstraintViolation();
+            return false;
+        }
+
+        return true;
+    }
+}

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

@@ -6,6 +6,7 @@ import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.tasks.ExecutableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.services.FlowService;
+import io.kestra.core.services.NamespaceService;
 import io.kestra.core.utils.ListUtils;
 import io.kestra.core.validations.FlowValidation;
 import io.micronaut.core.annotation.AnnotationValue;
@@ -52,6 +53,9 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
     @Inject
     private FlowService flowService;
 
+    @Inject
+    private NamespaceService namespaceService;
+
     @Override
     public boolean isValid(
         @Nullable Flow value,
@@ -67,7 +71,7 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
             violations.add("Flow id is a reserved keyword: " + value.getId() + ". List of reserved keywords: " + String.join(", ", RESERVED_FLOW_IDS));
         }
 
-        if (flowService.requireExistingNamespace(value.getTenantId(), value.getNamespace())) {
+        if (namespaceService.requireExistingNamespace(value.getTenantId(), value.getNamespace())) {
             violations.add("Namespace '" + value.getNamespace() + "' does not exist but is required to exist before a flow can be created in it.");
         }
 

core/src/main/java/io/kestra/plugin/core/condition/TimeBetween.java

@@ -79,20 +79,30 @@ public class TimeBetween extends Condition implements ScheduleCondition {
         RunContext runContext = conditionContext.getRunContext();
         Map<String, Object> variables = conditionContext.getVariables();
 
-        String dateRendered = runContext.render(date).as(String.class, variables).orElseThrow();
+        // cache must be skipped for date rendering as the value can change for each test
+        String dateRendered = runContext.render(date).skipCache().as(String.class, variables).orElseThrow();
         OffsetTime currentDate = DateUtils.parseZonedDateTime(dateRendered).toOffsetDateTime().toOffsetTime();
 
         OffsetTime beforeRendered = runContext.render(before).as(OffsetTime.class, variables).orElse(null);
         OffsetTime afterRendered = runContext.render(after).as(OffsetTime.class, variables).orElse(null);
-
+        
         if (beforeRendered != null && afterRendered != null) {
-            return currentDate.isAfter(afterRendered) && currentDate.isBefore(beforeRendered);
+            // Case 1: Normal range (e.g., 16:00 -> 20:00)
+            if (afterRendered.isBefore(beforeRendered)) {
+                return currentDate.isAfter(afterRendered) && currentDate.isBefore(beforeRendered);
+            // Case 2: Cross-midnight range (e.g., 22:00 -> 02:00)
+            } else {
+                return currentDate.isAfter(afterRendered) || currentDate.isBefore(beforeRendered);
+            }
+            
         } else if (beforeRendered != null) {
             return currentDate.isBefore(beforeRendered);
+            
         } else if (afterRendered != null) {
             return currentDate.isAfter(afterRendered);
+            
         } else {
-            throw new IllegalConditionEvaluation("Invalid condition with no before nor after");
+            throw new IllegalConditionEvaluation("Invalid condition: no 'before' or 'after' value defined");
         }
     }
 }

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Pie.java

@@ -1,7 +1,6 @@
 package io.kestra.plugin.core.dashboard.chart;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
-
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.dashboards.ColumnDescriptor;
@@ -21,34 +20,33 @@ import lombok.experimental.SuperBuilder;
 @EqualsAndHashCode
 @Schema(
     title = "Show proportions and distributions using pie charts."
-    )
+)
 @Plugin(
     examples = {
         @Example(
             title = "Display a pie chart with Executions per State.",
             full = true,
-            code = { """
+            code = """
                 charts:
-                    - id: executions_pie
-                      type: io.kestra.plugin.core.dashboard.chart.Pie
-                      chartOptions:
-                        displayName: Total Executions
-                        description: Total executions per state
-                      legend:
-                        enabled: true
-                        colorByColumn: state
-                      data:
-                        type: io.kestra.plugin.core.dashboard.data.Executions
-                        columns:
-                            state:
-                                field: STATE
-                            total:
-                                agg: COUNT
+                  - id: executions_pie
+                    type: io.kestra.plugin.core.dashboard.chart.Pie
+                    chartOptions:
+                      displayName: Total Executions
+                      description: Total executions per state
+                    legend:
+                      enabled: true
+                      colorByColumn: state
+                    data:
+                      type: io.kestra.plugin.core.dashboard.data.Executions
+                      columns:
+                        state:
+                          field: STATE
+                        total:
+                          agg: COUNT
                 """
-            }
         )
     }
-)    
+)
 public class Pie<F extends Enum<F>, D extends DataFilter<F, ? extends ColumnDescriptor<F>>> extends DataChart<PieOption, D> {
     @Override
     public Integer minNumberOfAggregations() {

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Table.java

@@ -1,7 +1,6 @@
 package io.kestra.plugin.core.dashboard.chart;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
-
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.dashboards.DataFilter;
@@ -21,33 +20,32 @@ import lombok.experimental.SuperBuilder;
 @EqualsAndHashCode
 @Schema(
     title = "Display structured data in a clear, sortable table."
-    )
+)
 @Plugin(
     examples = {
         @Example(
             title = "Display a table with Log counts for each level by Namespace.",
             full = true,
-            code = { """
+            code = """
                 charts:
-                    - id: table_logs
+                  - id: table_logs
                     type: io.kestra.plugin.core.dashboard.chart.Table
                     chartOptions:
-                        displayName: Log count by level for filtered namespace
+                      displayName: Log count by level for filtered namespace
                     data:
-                        type: io.kestra.plugin.core.dashboard.data.Logs
-                        columns:
-                            level:
-                                field: LEVEL
-                                count:
-                                    agg: COUNT
-                        where:
-                            - field: NAMESPACE
-                              type: IN
-                              values:
-                                - dev_graph
-                                - prod_graph
+                      type: io.kestra.plugin.core.dashboard.data.Logs
+                      columns:
+                        level:
+                          field: LEVEL
+                          count:
+                           agg: COUNT
+                      where:
+                        - field: NAMESPACE
+                          type: IN
+                          values:
+                            - dev_graph
+                            - prod_graph
                 """
-            }
         )
     }
 )

core/src/main/java/io/kestra/plugin/core/dashboard/chart/TimeSeries.java

@@ -1,7 +1,6 @@
 package io.kestra.plugin.core.dashboard.chart;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
-
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.dashboards.DataFilter;
@@ -23,42 +22,41 @@ import lombok.experimental.SuperBuilder;
 @TimeSeriesChartValidation
 @Schema(
     title = "Track trends over time with dynamic time series charts."
-    )
+)
 @Plugin(
     examples = {
         @Example(
             title = "Display a chart with Executions over the last week.",
             full = true,
-            code = { """
+            code = """
                 charts:
                   - id: executions_timeseries
                     type: io.kestra.plugin.core.dashboard.chart.TimeSeries
                     chartOptions:
-                        displayName: Total Executions
-                        description: Executions last week
-                        legend:
-                            enabled: true
-                        column: date
-                        colorByColumn: state
+                      displayName: Total Executions
+                      description: Executions last week
+                      legend:
+                        enabled: true
+                      column: date
+                      colorByColumn: state
                     data:
-                        type: io.kestra.plugin.core.dashboard.data.Executions
-                        columns:
-                            date:
-                                field: START_DATE
-                                displayName: Date
-                            state:
-                                field: STATE
-                            total:
-                                displayName: Executions
-                                agg: COUNT
-                                graphStyle: BARS
-                            duration:
-                                displayName: Duration
-                                field: DURATION
-                                agg: SUM
-                                graphStyle: LINES
+                      type: io.kestra.plugin.core.dashboard.data.Executions
+                      columns:
+                        date:
+                          field: START_DATE
+                          displayName: Date
+                        state:
+                          field: STATE
+                        total:
+                          displayName: Executions
+                          agg: COUNT
+                          graphStyle: BARS
+                        duration:
+                          displayName: Duration
+                          field: DURATION
+                          agg: SUM
+                          graphStyle: LINES
                 """
-            }
         )
     }
 )

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

@@ -30,28 +30,27 @@ import lombok.experimental.SuperBuilder;
         @Example(
             title = "Display a chart with a Executions per Namespace broken out by State.",
             full = true,
-            code = { """
-            charts:
-              - id: executions_per_namespace_bars
-                type: io.kestra.plugin.core.dashboard.chart.Bar
-                chartOptions:
-                  displayName: Executions (per namespace)
-                  description: Executions count per namespace
-                  legend:
-                    enabled: true
-                  column: namespace
-                data
-                  type: io.kestra.plugin.core.dashboard.data.Executions
-                  columns:
-                    namespace:
-                      field: NAMESPACE
-                    state:
-                      field: STATE
-                    total:
-                      displayName: Executions
-                      agg: COUNT
-            """
-          }
+            code = """
+                charts:
+                  - id: executions_per_namespace_bars
+                    type: io.kestra.plugin.core.dashboard.chart.Bar
+                    chartOptions:
+                      displayName: Executions (per namespace)
+                      description: Executions count per namespace
+                      legend:
+                        enabled: true
+                      column: namespace
+                    data
+                      type: io.kestra.plugin.core.dashboard.data.Executions
+                      columns:
+                        namespace:
+                          field: NAMESPACE
+                        state:
+                          field: STATE
+                        total:
+                          displayName: Executions
+                          agg: COUNT
+                """
         )
     }
 )

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

@@ -30,26 +30,25 @@ import lombok.experimental.SuperBuilder;
         @Example(
             title = "Display a chart with executions in success in a given namespace.",
             full = true,
-            code = { """
-            charts:
-              - id: kpi_success_ratio
-                type: io.kestra.plugin.core.dashboard.chart.KPI
-                chartOptions:
-                  displayName: Success Ratio
-                  numberType: PERCENTAGE
-                  width: 3
-                data:
-                  type: io.kestra.plugin.core.dashboard.data.ExecutionsKPI
-                  columns:
-                    field: ID
-                    agg: COUNT
-                  numerator:
-                    - type: IN
-                      field: STATE
-                      values:
-                        - SUCCESS
-            """
-          }
+            code = """
+                charts:
+                  - id: kpi_success_ratio
+                    type: io.kestra.plugin.core.dashboard.chart.KPI
+                    chartOptions:
+                      displayName: Success Ratio
+                      numberType: PERCENTAGE
+                      width: 3
+                    data:
+                      type: io.kestra.plugin.core.dashboard.data.ExecutionsKPI
+                      columns:
+                        field: ID
+                        agg: COUNT
+                      numerator:
+                        - type: IN
+                          field: STATE
+                          values:
+                            - SUCCESS
+                """
         )
     }
 )

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

@@ -27,19 +27,18 @@ import lombok.experimental.SuperBuilder;
         @Example(
             title = "Display a chart with a list of Flows.",
             full = true,
-            code = { """
-            charts:
-              - id: list_flows
-                type: io.kestra.plugin.core.dashboard.chart.Table
-                data:
-                  type: io.kestra.plugin.core.dashboard.data.Flows
-                  columns:
-                    namespace:
-                      field: NAMESPACE
-                    id:
-                      field: ID
-            """
-          }
+            code = """
+                charts:
+                  - id: list_flows
+                    type: io.kestra.plugin.core.dashboard.chart.Table
+                    data:
+                      type: io.kestra.plugin.core.dashboard.data.Flows
+                      columns:
+                        namespace:
+                          field: NAMESPACE
+                        id:
+                          field: ID
+                """
         )
     }
 )

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

@@ -28,17 +28,16 @@ import lombok.experimental.SuperBuilder;
         @Example(
             title = "Display count of Flows.",
             full = true,
-            code = { """
-            charts:
-              - id: kpi
+            code = """
+                charts:
+                  - id: kpi
                     type: io.kestra.plugin.core.dashboard.chart.KPI
                     data:
                       type: io.kestra.plugin.core.dashboard.data.FlowsKPI
                       columns:
                         field: ID
                         agg: COUNT
-            """
-          }
+                """
         )
     }
 )

core/src/main/java/io/kestra/plugin/core/debug/Return.java

@@ -48,11 +48,11 @@ import java.util.Optional;
                 id: compute_header
                 type: io.kestra.plugin.core.debug.Return
                 format: >-
-                    {%- if inputs.token is not empty -%}
-                    Bearer {{ inputs.token }}
-                    {%- elseif inputs.username is not empty and inputs.password is not empty -%}
-                    Basic {{ (inputs.username + ':' + inputs.password) | base64encode }}
-                    {%- endif -%}
+                  {%- if inputs.token is not empty -%}
+                  Bearer {{ inputs.token }}
+                  {%- elseif inputs.username is not empty and inputs.password is not empty -%}
+                  Basic {{ (inputs.username + ':' + inputs.password) | base64encode }}
+                  {%- endif -%}
                 """
         )
     },

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

@@ -54,8 +54,8 @@ import java.util.concurrent.atomic.AtomicInteger;
                 "  - id: fail\n" +
                 "    type: io.kestra.plugin.core.execution.Assert\n" +
                 "    conditions:\n" +
-                "     - \"{{ inputs.param == 'ok' }}\"\n" +
-                "     - \"{{ 1 + 1 == 3 }}\"\n"
+                "      - \"{{ inputs.param == 'ok' }}\"\n" +
+                "      - \"{{ 1 + 1 == 3 }}\"\n"
             }
         )
     },

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

@@ -13,7 +13,6 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.repositories.ExecutionRepositoryInterface;
 import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.services.FlowService;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
@@ -127,14 +126,13 @@ public class Count extends Task implements RunnableTask<Count.Output> {
         var flowInfo = runContext.flowInfo();
 
         // check that all flows are allowed
-        FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
         if (flows != null) {
-            flows.forEach(flow -> flowService.checkAllowedNamespace(flowInfo.tenantId(), flow.getNamespace(), flowInfo.tenantId(), flowInfo.namespace()));
+            flows.forEach(flow -> runContext.acl().allowNamespace(flow.getNamespace()).check());
         }
 
         if (namespaces != null) {
             var renderedNamespaces = runContext.render(this.namespaces).asList(String.class);
-            renderedNamespaces.forEach(namespace -> flowService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace()));
+            renderedNamespaces.forEach(namespace -> runContext.acl().allowNamespace(namespace).check());
         }
 
         List<ExecutionCount> executionCounts = executionRepository.executionCounts(

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

@@ -105,7 +105,7 @@ import lombok.experimental.SuperBuilder;
                 url: "{{ secret('SLACK_WEBHOOK') }}"
                 payload: |
                 {
-                    "text": "Failure alert for flow `{{ flow.namespace }}.{{ flow.id }}` with ID `{{ execution.id }}`. Here is a bit more context about why the execution failed: `{{ errorLogs()[0]['message'] }}`"
+                  "text": "Failure alert for flow `{{ flow.namespace }}.{{ flow.id }}` with ID `{{ execution.id }}`. Here is a bit more context about why the execution failed: `{{ errorLogs()[0]['message'] }}`"
                 }
             """
         )

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

@@ -9,7 +9,6 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.services.ExecutionService;
-import io.kestra.core.services.FlowService;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
@@ -34,10 +33,10 @@ import java.util.List;
             code = {
                 "endDate: \"{{ now() | dateAdd(-1, 'MONTHS') }}\"",
                 "states: ",
-                " - KILLED",
-                " - FAILED",
-                " - WARNING",
-                " - SUCCESS"
+                "  - KILLED",
+                "  - FAILED",
+                "  - WARNING",
+                "  - SUCCESS"
             }
         )
     },
@@ -113,15 +112,14 @@ public class PurgeExecutions extends Task implements RunnableTask<PurgeExecution
     @Override
     public PurgeExecutions.Output run(RunContext runContext) throws Exception {
         ExecutionService executionService = ((DefaultRunContext)runContext).getApplicationContext().getBean(ExecutionService.class);
-        FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
 
         // validate that this namespace is authorized on the target namespace / all namespaces
         var flowInfo = runContext.flowInfo();
         String renderedNamespace = runContext.render(this.namespace).as(String.class).orElse(null);
         if (renderedNamespace == null){
-            flowService.checkAllowedAllNamespaces(flowInfo.tenantId(), flowInfo.tenantId(), flowInfo.namespace());
+            runContext.acl().allowAllNamespaces().check();
         } else if (!renderedNamespace.equals(flowInfo.namespace())) {
-            flowService.checkAllowedNamespace(flowInfo.tenantId(), renderedNamespace, flowInfo.tenantId(), flowInfo.namespace());
+            runContext.acl().allowNamespace(renderedNamespace).check();
         }
 
         ExecutionService.PurgeResult purgeResult = executionService.purge(

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

@@ -1,11 +1,5 @@
 package io.kestra.plugin.core.flow;
 
-import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
-import lombok.experimental.SuperBuilder;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
@@ -17,6 +11,12 @@ import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.VoidOutput;
 import io.kestra.core.runners.FlowableUtils;
 import io.kestra.core.runners.RunContext;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
 
 import java.util.List;
 import java.util.Optional;
@@ -60,24 +60,23 @@ import java.util.Optional;
                 namespace: company.team
 
                 tasks:
-                - id: allow_failure
+                  - id: allow_failure
                     type: io.kestra.plugin.core.flow.AllowFailure
                     tasks:
-                    - id: fail_silently
+                      - id: fail_silently
                         type: io.kestra.plugin.scripts.shell.Commands
                         taskRunner:
                         type: io.kestra.plugin.core.runner.Process
                         commands:
-                        - exit 1
+                          - exit 1
 
-                - id: print_to_console
+                  - id: print_to_console
                     type: io.kestra.plugin.scripts.shell.Commands
                     taskRunner:
                     type: io.kestra.plugin.core.runner.Process
                     commands:
-                    - echo "this will run since previous failure was allowed ✅"
-
-            """
+                      - echo "this will run since previous failure was allowed ✅"
+                """
         )
     },
     aliases = "io.kestra.core.tasks.flows.AllowFailure"

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

@@ -8,6 +8,7 @@ import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.NextTaskRun;
 import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.flows.State;
 import io.kestra.core.models.hierarchies.GraphCluster;
 import io.kestra.core.models.hierarchies.RelationType;
 import io.kestra.core.models.property.Property;
@@ -15,6 +16,7 @@ import io.kestra.core.models.tasks.*;
 import io.kestra.core.runners.FlowableUtils;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.utils.GraphUtils;
+import io.kestra.core.utils.ListUtils;
 import io.kestra.core.validations.DagTaskValidation;
 import io.micronaut.core.annotation.Introspected;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -176,6 +178,22 @@ public class Dag extends Task implements FlowableTask<VoidOutput> {
         );
     }
 
+    @Override
+    public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        List<ResolvedTask> childTasks = this.childTasks(runContext, parentTaskRun);
+
+        return FlowableUtils.resolveSequentialState(
+            execution,
+            childTasks,
+            FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),
+            FlowableUtils.resolveTasks(this.getFinally(), parentTaskRun),
+            parentTaskRun,
+            runContext,
+            this.isAllowFailure(),
+            this.isAllowWarning()
+        );
+    }
+
     public List<String> dagCheckNotExistTask(List<DagTask> taskDepends) {
         List<String> dependenciesIds = taskDepends
             .stream()

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

@@ -163,15 +163,9 @@ public class EachParallel extends Parallel implements FlowableTask<VoidOutput> {
 
     @Override
     public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
-        List<ResolvedTask> childTasks = ListUtils.emptyOnNull(this.childTasks(runContext, parentTaskRun)).stream()
-            .filter(resolvedTask -> !resolvedTask.getTask().getDisabled())
-            .toList();
+        List<ResolvedTask> childTasks = this.childTasks(runContext, parentTaskRun);
 
-        if (childTasks.isEmpty()) {
-            return Optional.of(State.Type.SUCCESS);
-        }
-
-        return FlowableUtils.resolveState(
+        return FlowableUtils.resolveSequentialState(
             execution,
             childTasks,
             FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),

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

@@ -127,14 +127,9 @@ public class EachSequential extends Sequential implements FlowableTask<VoidOutpu
 
     @Override
     public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
-        List<ResolvedTask> childTasks = ListUtils.emptyOnNull(this.childTasks(runContext, parentTaskRun)).stream()
-            .filter(resolvedTask -> !resolvedTask.getTask().getDisabled())
-            .toList();
-        if (childTasks.isEmpty()) {
-            return Optional.of(State.Type.SUCCESS);
-        }
+        List<ResolvedTask> childTasks = this.childTasks(runContext, parentTaskRun);
 
-        return FlowableUtils.resolveState(
+        return FlowableUtils.resolveSequentialState(
             execution,
             childTasks,
             FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),

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

@@ -36,23 +36,26 @@ import java.util.Optional;
     description = """
         You can control how many task groups are executed concurrently by setting the `concurrencyLimit` property. \
 
-        - If you set the `concurrencyLimit` property to `0`, Kestra will execute all task groups concurrently for all values. \
+        - A `concurrencyLimit` of `0` means no limit — all task groups run in parallel. \
 
-        - If you set the `concurrencyLimit` property to `1`, Kestra will execute each task group one after the other starting with the task group for the first value in the list. \
+        - A `concurrencyLimit` of `1` means full serialization — only one task group runs at a time, in order. \
+
+        - A `concurrencyLimit` greater than `1` allows up to that number of task groups to run in parallel. \
 
 
         Regardless of the `concurrencyLimit` property, the `tasks` will run one after the other — to run those in parallel, wrap them in a [Parallel](https://kestra.io/plugins/core/tasks/flow/io.kestra.plugin.core.flow.parallel) task as shown in the last example below (_see the flow `parallel_tasks_example`_). \
 
 
-        The `values` should be defined as a JSON string or an array, e.g. a list of string values `["value1", "value2"]` or a list of key-value pairs `[{"key": "value1"}, {"key": "value2"}]`.\s
+        The `values` can be defined as a JSON string or an array, e.g. a list of string values `["value1", "value2"]` or a list of key-value pairs `[{"key": "value1"}, {"key": "value2"}]`.\s
 
 
-        You can access the current iteration value using the variable `{{ taskrun.value }}` \
-        or `{{ parent.taskrun.value }}` if you are in a nested child task. You can access the batch or iteration number with `{{ taskrun.iteration }}`. \
+        Access the current iteration value using `{{ taskrun.value }}` \
+        or `{{ parent.taskrun.value }}` when inside a nested child task. \
+        The iteration number is available via `{{ taskrun.iteration }}`. \
 
 
         If you need to execute more than 2-5 tasks for each value, we recommend triggering a subflow for each value for better performance and modularity. \
-        Check the [flow best practices documentation](https://kestra.io/docs/best-practices/flows) for more details."""
+        See the [flow best practices documentation](https://kestra.io/docs/best-practices/flows) for more details."""
 )
 @Plugin(
     examples = {
@@ -210,12 +213,14 @@ public class ForEach extends Sequential implements FlowableTask<VoidOutput> {
     @NotNull
     @Builder.Default
     @Schema(
-        title = "The number of concurrent task groups for each value in the `values` array",
-        description = """
-        If you set the `concurrencyLimit` property to 0, Kestra will execute all task groups concurrently for all values (zero limits!). \
+      title = "The number of concurrent task groups for each value in the `values` array",
+      description = """
+        A `concurrencyLimit` of 0 means no limit — all task groups run in parallel.
 
+        A `concurrencyLimit` of 1 means full serialization — only one task group runs at a time, in order.
 
-        If you set the `concurrencyLimit` property to 1, Kestra will execute each task group one after the other starting with the first value in the list (limit concurrency to one task group that can be actively running at any time)."""
+        A `concurrencyLimit` greater than 1 allows up to the specified number of task groups to run in parallel.
+        """
     )
     @PluginProperty
     private final Integer concurrencyLimit = 1;
@@ -245,15 +250,9 @@ public class ForEach extends Sequential implements FlowableTask<VoidOutput> {
 
     @Override
     public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
-        List<ResolvedTask> childTasks = ListUtils.emptyOnNull(this.childTasks(runContext, parentTaskRun)).stream()
-            .filter(resolvedTask -> !resolvedTask.getTask().getDisabled())
-            .toList();
+        List<ResolvedTask> childTasks = this.childTasks(runContext, parentTaskRun);
 
-        if (childTasks.isEmpty()) {
-            return Optional.of(State.Type.SUCCESS);
-        }
-
-        return FlowableUtils.resolveState(
+        return FlowableUtils.resolveSequentialState(
             execution,
             childTasks,
             FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),

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

@@ -1,19 +1,17 @@
 package io.kestra.plugin.core.flow;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import io.kestra.core.models.annotations.PluginProperty;
-import io.kestra.core.models.property.Property;
-import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.*;
-import lombok.experimental.SuperBuilder;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.NextTaskRun;
 import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.flows.State;
 import io.kestra.core.models.hierarchies.GraphCluster;
 import io.kestra.core.models.hierarchies.RelationType;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
@@ -21,12 +19,16 @@ import io.kestra.core.models.tasks.VoidOutput;
 import io.kestra.core.runners.FlowableUtils;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.utils.GraphUtils;
-
-import java.util.List;
-import java.util.stream.Stream;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotEmpty;
 import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Stream;
 
 @SuperBuilder
 @ToString
@@ -42,8 +44,8 @@ import jakarta.validation.constraints.NotNull;
         @Example(
             full = true,
             title = """
-            Run tasks in parallel
-            """,
+                Run tasks in parallel
+                """,
             code = """
                 id: parallel
                 namespace: company.team
@@ -68,38 +70,38 @@ import jakarta.validation.constraints.NotNull;
         @Example(
             full = true,
             title = """
-            Run two sequences in parallel
-            """,
+                Run two sequences in parallel
+                """,
             code = """
                 id: parallel_sequences
                 namespace: company.team
 
                 tasks:
-                - id: parallel
+                  - id: parallel
                     type: io.kestra.plugin.core.flow.Parallel
                     tasks:
-                    - id: sequence1
+                      - id: sequence1
                         type: io.kestra.plugin.core.flow.Sequential
                         tasks:
-                        - id: task1
+                          - id: task1
                             type: io.kestra.plugin.core.debug.Return
                             format: "{{ task.id }}"
 
-                        - id: task2
+                          - id: task2
                             type: io.kestra.plugin.core.debug.Return
                             format: "{{ task.id }}"
 
-                    - id: sequence2
+                      - id: sequence2
                         type: io.kestra.plugin.core.flow.Sequential
                         tasks:
-                        - id: task3
+                          - id: task3
                             type: io.kestra.plugin.core.debug.Return
                             format: "{{ task.id }}"
 
-                        - id: task4
+                          - id: task4
                             type: io.kestra.plugin.core.debug.Return
                             format: "{{ task.id }}"
-            """
+                """
         )
     },
     aliases = "io.kestra.core.tasks.flows.Parallel"
@@ -176,4 +178,20 @@ public class Parallel extends Task implements FlowableTask<VoidOutput> {
             runContext.render(this.concurrent).as(Integer.class).orElseThrow()
         );
     }
+
+    @Override
+    public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        List<ResolvedTask> childTasks = this.childTasks(runContext, parentTaskRun);
+
+        return FlowableUtils.resolveSequentialState(
+            execution,
+            childTasks,
+            FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),
+            FlowableUtils.resolveTasks(this.getFinally(), parentTaskRun),
+            parentTaskRun,
+            runContext,
+            this.isAllowFailure(),
+            this.isAllowWarning()
+        );
+    }
 }