chore(core): localize to languages other than english (#13280)

Extended localization support by adding translations for multiple languages using English as the base. This enhances accessibility and usability for non-English-speaking users while keeping English as the source reference.

Co-authored-by: GitHub Action <actions@github.com>

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

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

.github/workflows/codeql-analysis.yml

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

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

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

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

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

.github/workflows/main-build.yml

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

.github/workflows/pull-request.yml

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

.github/workflows/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

@@ -7,7 +7,7 @@ buildscript {
     }
 
     dependencies {
-        classpath "net.e175.klaus:zip-prefixer:0.3.1"
+        classpath "net.e175.klaus:zip-prefixer:0.4.0"
     }
 }
 
@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "7.0.1.6134"
+    id "org.sonarqube" version "7.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"
 

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,8 +1,10 @@
 package io.kestra.cli.commands.migrations.metadata;
 
 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;
@@ -15,11 +17,13 @@ import jakarta.inject.Singleton;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.file.Path;
 import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static io.kestra.core.utils.Rethrow.throwConsumer;
@@ -36,6 +40,9 @@ public class MetadataMigrationService {
     @Inject
     private KvMetadataRepositoryInterface kvMetadataRepository;
 
+    @Inject
+    private NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository;
+
     @Inject
     private StorageInterface storageInterface;
 
@@ -49,7 +56,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 +84,35 @@ 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)));
+            String prefix = prefixFunction.apply(namespace);
+            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 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/test/java/io/kestra/cli/commands/migrations/metadata/NsFilesMetadataMigrationCommandTest.java

@@ -0,0 +1,145 @@
+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);
+        }
+    }
+
+    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/services/FileChangedEventListenerTest.java

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

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

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

core/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/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

@@ -7,7 +7,6 @@ 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;
@@ -215,8 +214,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/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)
@@ -173,7 +177,7 @@ public class RunContextFactory {
         RunContextLogger runContextLogger = new RunContextLogger();
         return newBuilder()
             .withLogger(runContextLogger)
-            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forFlow(flow), storageInterface, flowService))
+            .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forFlow(flow), storageInterface, namespaceService, namespaceFactory))
             .withVariables(variables)
             .withSecretInputs(secretInputsFromFlow(flow))
             .build();
@@ -212,7 +216,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/RunnerUtils.java

@@ -160,7 +160,7 @@ public class RunnerUtils {
             executionEmitter.run();
 
             if (duration == null) {
-                Await.until(() -> receive.get() != null, Duration.ofMillis(10));
+                Await.untilWithSleepInterval(() -> receive.get() != null, Duration.ofMillis(10));
             } else {
                 Await.until(() -> receive.get() != null, Duration.ofMillis(10), duration);
             }

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,7 +23,7 @@ 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;

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/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);
     }
 
     /**

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/utils/Await.java

@@ -10,10 +10,14 @@ public class Await {
     private static final Duration defaultSleep = Duration.ofMillis(100);
 
     public static void until(BooleanSupplier condition) {
-        Await.until(condition, null);
+        Await.untilWithSleepInterval(condition, null);
     }
 
-    public static void until(BooleanSupplier condition, Duration sleep) {
+    public static void untilWithTimeout(BooleanSupplier condition, Duration timeout) throws TimeoutException {
+        until(condition, null, timeout);
+    }
+
+    public static void untilWithSleepInterval(BooleanSupplier condition, Duration sleep) {
         if (sleep == null) {
             sleep = defaultSleep;
         }
@@ -74,10 +78,10 @@ public class Await {
         return result.get();
     }
 
-    public static <T> T until(Supplier<T> supplier, Duration sleep) {
+    public static <T> T untilWithSleepInterval(Supplier<T> supplier, Duration sleep) {
         AtomicReference<T> result = new AtomicReference<>();
 
-        Await.until(untilSupplier(supplier, result), sleep);
+        Await.untilWithSleepInterval(untilSupplier(supplier, result), sleep);
 
         return result.get();
     }

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

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

@@ -42,7 +42,7 @@ import java.util.*;
 @Schema(
     title = "Pause the current execution and wait for approval (either by humans or other automated processes).",
     description = "All tasks downstream from the Pause task will be put on hold until the execution is manually resumed from the UI.\n\n" +
-      "The Execution will be in a Paused state, and you can either manually resume it by clicking on the \"Resume\" button in the UI or by calling the POST API endpoint `/api/v1/executions/{executionId}/resume`. The execution can also be resumed automatically after the `pauseDuration`."
+        "The Execution will be in a Paused state, and you can either manually resume it by clicking on the \"Resume\" button in the UI or by calling the POST API endpoint `/api/v1/executions/{executionId}/resume`. The execution can also be resumed automatically after the `pauseDuration`."
 )
 @Plugin(
     examples = {
@@ -185,7 +185,7 @@ public class Pause extends Task implements FlowableTask<Pause.Output> {
     )
     @NotNull
     @Builder.Default
-    private Property<Behavior> behavior = Property.ofValue(Behavior.RESUME);
+    protected Property<Behavior> behavior = Property.ofValue(Behavior.RESUME);
 
     @Valid
     @Schema(

core/src/main/java/io/kestra/plugin/core/flow/Sequential.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.AbstractGraph;
 import io.kestra.core.models.hierarchies.GraphCluster;
 import io.kestra.core.models.hierarchies.RelationType;
@@ -23,6 +24,7 @@ import lombok.experimental.SuperBuilder;
 import jakarta.validation.Valid;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Stream;
 
 @SuperBuilder
@@ -113,6 +115,22 @@ public class Sequential extends Task implements FlowableTask<VoidOutput> {
         return FlowableUtils.resolveTasks(this.getTasks(), parentTaskRun);
     }
 
+    @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()
+        );
+    }
+
     @Override
     public List<NextTaskRun> resolveNexts(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
         return FlowableUtils.resolveSequentialNexts(

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

@@ -141,12 +141,12 @@ import jakarta.validation.constraints.NotNull;
                       - id: first
                         type: io.kestra.plugin.scripts.shell.Commands
                         commands:
-                        - 'echo "{{ taskrun.id }}" > {{ workingDir }}/stay.txt'
+                          - 'echo "{{ taskrun.id }}" > {{ workingDir }}/stay.txt'
                       - id: second
                         type: io.kestra.plugin.scripts.shell.Commands
                         commands:
-                        - |
-                          echo '::{"outputs": {"stay":"'$(cat {{ workingDir }}/stay.txt)'"}}::''
+                          - |
+                            echo '::{"outputs": {"stay":"'$(cat {{ workingDir }}/stay.txt)'"}}::''
                 """
         ),
         @Example(

core/src/main/java/io/kestra/plugin/core/http/Request.java

@@ -34,10 +34,11 @@ import java.util.OptionalInt;
 @Schema(
     title = "Make an HTTP API request to a specified URL and store the response as an output.",
     description = """
-                  This task makes an API call to a specified URL of an HTTP server and stores the response as an output.
-                  Kestra offers hundreds of plugins. Before using the generic HTTP task, check if a dedicated plugin fits your use case — it's recommended to use plugins first and only fall back to HTTP when needed.
-                  By default, the maximum length of the response is limited to 10MB, but it can be increased to at most 2GB by using the `options.maxContentLength` property.
-                  Note that the response is added as an output of the task. If you need to process large API payloads, we recommend using the `Download` task instead."""
+        This task makes an API call to a specified URL of an HTTP server and stores the response as an output.
+        Kestra offers hundreds of plugins. Before using the generic HTTP task, check if a dedicated plugin fits your use case — it's recommended to use plugins first and only fall back to HTTP when needed.
+        By default, the maximum length of the response is limited to 10MB, but it can be increased to at most 2GB by using the `options.maxContentLength` property.
+        Note that the response is added as an output of the task. If you need to process large API payloads, we recommend using the `Download` task instead.
+        """
 )
 @Plugin(
     examples = {
@@ -277,42 +278,45 @@ import java.util.OptionalInt;
                 """
         ),
         @Example(
-          title = "Send a multiline JSON message using HTTP POST request and inputs with a pebble expression. We recommend this method to avoid JSON string interpolation",
-          full = true,
-          code = """
-              id: http_multiline_json
-              namespace: company.team
+            title = "Send a multiline JSON message using HTTP POST request and inputs with a pebble expression. We recommend this method to avoid JSON string interpolation",
+            full = true,
+            code = """
+                id: http_multiline_json
+                namespace: company.team
 
-              inputs:
-                - id: title
-                  type: STRING
-                  defaults: This is the title of the request
-                - id: message
-                  type: STRING
-                  defaults: |-
-                    This is my long
-                    multiline message.
-                - id: priority
-                  type: INT
-                  defaults: 5
+                inputs:
+                  - id: title
+                    type: STRING
+                    defaults: This is the title of the request
+                  - id: message
+                    type: STRING
+                    defaults: |-
+                      This is my long
+                      multiline message.
+                  - id: priority
+                    type: INT
+                    defaults: 5
 
-              tasks:
-                - id: send
-                  type: io.kestra.plugin.core.http.Request
-                  uri: "https://reqres.in/api/test-request"
-                  method: "POST"
-                  body: |
-                    {{ {
-                      "title": inputs.title,
-                      "message": inputs.message,
-                      "priority": inputs.priority
-                    } }}
-              """
-      )
+                tasks:
+                  - id: send
+                    type: io.kestra.plugin.core.http.Request
+                    uri: "https://reqres.in/api/test-request"
+                    method: "POST"
+                    body: |
+                      {{ {
+                        "title": inputs.title,
+                        "message": inputs.message,
+                        "priority": inputs.priority
+                      } }}
+                """
+        )
     },
     aliases = "io.kestra.plugin.fs.http.Request"
 )
 public class Request extends AbstractHttp implements RunnableTask<Request.Output> {
+
+    private static final int MAX_OUTPUT_BODY_BYTES = 19 * 1024 * 1024; // ~19MB safety margin
+
     @Builder.Default
     @Schema(
         title = "If true, the HTTP response body will be automatically encrypted and decrypted in the outputs, provided that encryption is configured in your Kestra configuration.",
@@ -329,7 +333,8 @@ public class Request extends AbstractHttp implements RunnableTask<Request.Output
             String body = null;
 
             if (response.getBody() != null) {
-                body = IOUtils.toString(ArrayUtils.toPrimitive(response.getBody()), StandardCharsets.UTF_8.name());
+                byte[] bytes = getResponseBytes(response);
+                body = IOUtils.toString(bytes, StandardCharsets.UTF_8.name());
             }
 
             // check that the string is a valid Unicode string
@@ -346,6 +351,17 @@ public class Request extends AbstractHttp implements RunnableTask<Request.Output
         }
     }
 
+    private static byte[] getResponseBytes(HttpResponse<Byte[]> response) {
+        byte[] bytes = ArrayUtils.toPrimitive(response.getBody());
+        if (bytes.length > MAX_OUTPUT_BODY_BYTES) {
+            throw new IllegalArgumentException(
+                "Response body is too large to store in task outputs (" + bytes.length + " bytes > max " + MAX_OUTPUT_BODY_BYTES + " bytes). " +
+                    "Use io.kestra.plugin.core.http.Download to fetch large payloads as files instead."
+            );
+        }
+        return bytes;
+    }
+
     public Output output(RunContext runContext, HttpRequest request, HttpResponse<Byte[]> response, String body) throws GeneralSecurityException, URISyntaxException, IOException, IllegalVariableEvaluationException {
         boolean encrypt = runContext.render(this.encryptBody).as(Boolean.class).orElseThrow();
         return Output.builder()

core/src/main/java/io/kestra/plugin/core/kv/Delete.java

@@ -5,9 +5,7 @@ import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
-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.Builder;
@@ -65,9 +63,7 @@ public class Delete extends Task implements RunnableTask<Delete.Output> {
     @Override
     public Output run(RunContext runContext) throws Exception {
         String renderedNamespace = runContext.render(this.namespace).as(String.class).orElseThrow();
-
-        FlowService flowService = ((DefaultRunContext) runContext).getApplicationContext().getBean(FlowService.class);
-        flowService.checkAllowedNamespace(runContext.flowInfo().tenantId(), renderedNamespace, runContext.flowInfo().tenantId(), runContext.flowInfo().namespace());
+        runContext.acl().allowNamespace(renderedNamespace).check();
 
         String renderedKey = runContext.render(this.key).as(String.class).orElseThrow();
 

core/src/main/java/io/kestra/plugin/core/kv/Get.java

@@ -8,7 +8,6 @@ import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.services.FlowService;
 import io.kestra.core.services.KVStoreService;
 import io.kestra.core.storages.kv.KVValue;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -82,8 +81,7 @@ public class Get extends Task implements RunnableTask<Get.Output> {
         if (Objects.equals(renderedNamespace, flowNamespace)) {
             value = getValueWithInheritance(runContext, flowNamespace, renderedKey);
         } else {
-            FlowService flowService = ((DefaultRunContext) runContext).getApplicationContext().getBean(FlowService.class);
-            flowService.checkAllowedNamespace(runContext.flowInfo().tenantId(), renderedNamespace, runContext.flowInfo().tenantId(), runContext.flowInfo().namespace());
+            runContext.acl().allowNamespace(renderedNamespace).check();
             value = runContext.namespaceKv(renderedNamespace).getValue(renderedKey);
         }
 

core/src/main/java/io/kestra/plugin/core/kv/GetKeys.java

@@ -5,9 +5,7 @@ import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
-import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.services.FlowService;
 import io.kestra.core.storages.kv.KVEntry;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
@@ -18,7 +16,6 @@ import lombok.experimental.SuperBuilder;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.*;
-import java.util.function.Function;
 import java.util.function.Predicate;
 
 @Slf4j
@@ -63,9 +60,7 @@ public class GetKeys extends Task implements RunnableTask<GetKeys.Output> {
     @Override
     public Output run(RunContext runContext) throws Exception {
         String renderedNamespace = runContext.render(this.namespace).as(String.class).orElse(null);
-
-        FlowService flowService = ((DefaultRunContext) runContext).getApplicationContext().getBean(FlowService.class);
-        flowService.checkAllowedNamespace(runContext.flowInfo().tenantId(), renderedNamespace, runContext.flowInfo().tenantId(), runContext.flowInfo().namespace());
+        runContext.acl().allowNamespace(renderedNamespace).check();
 
         String renderedPrefix = runContext.render(this.prefix).as(String.class).orElse(null);
         Predicate<String> filter = renderedPrefix == null ? key -> true : key -> key.startsWith(renderedPrefix);

core/src/main/java/io/kestra/plugin/core/kv/Key.java

@@ -17,7 +17,7 @@ import java.util.List;
 @SuperBuilder
 @Getter
 @NoArgsConstructor
-public class Key extends PurgeBehavior {
+public class Key extends KvPurgeBehavior {
     @NotNull
     @JsonInclude
     @Builder.Default

core/src/main/java/io/kestra/plugin/core/kv/KvPurgeBehavior.java

@@ -21,7 +21,7 @@ import java.util.List;
 @NoArgsConstructor
 @SuperBuilder
 @Introspected
-public abstract class PurgeBehavior {
+public abstract class KvPurgeBehavior {
     abstract public String getType();
 
     protected abstract List<KVEntry> entriesToPurge(KVStore kvStore) throws IOException;

core/src/main/java/io/kestra/plugin/core/kv/PurgeKV.java

@@ -1,6 +1,5 @@
 package io.kestra.plugin.core.kv;
 
-
 import com.cronutils.utils.VisibleForTesting;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.ValidationErrorException;
@@ -12,10 +11,10 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.services.FlowService;
 import io.kestra.core.storages.kv.KVEntry;
 import io.kestra.core.storages.kv.KVStore;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.plugin.core.purge.PurgeTask;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.Valid;
 import lombok.Builder;
@@ -46,7 +45,7 @@ import java.util.concurrent.atomic.AtomicLong;
             code = """
                 id: purge_kv_store
                 namespace: system
-                
+
                 tasks:
                   - id: purge_kv
                     type: io.kestra.plugin.core.kv.PurgeKV
@@ -58,7 +57,7 @@ import java.util.concurrent.atomic.AtomicLong;
         )
     }
 )
-public class PurgeKV extends Task implements RunnableTask<PurgeKV.Output> {
+public class PurgeKV extends Task implements PurgeTask<KVEntry>, RunnableTask<PurgeKV.Output> {
     @Schema(
         title = "Key pattern, e.g. 'AI_*'",
         description = "Delete only keys matching the glob pattern."
@@ -83,7 +82,7 @@ public class PurgeKV extends Task implements RunnableTask<PurgeKV.Output> {
     )
     @Builder.Default
     @Valid
-    private Property<PurgeBehavior> behavior = Property.ofValue(Key.builder().expiredOnly(true).build());
+    private Property<KvPurgeBehavior> behavior = Property.ofValue(Key.builder().expiredOnly(true).build());
 
     @Schema(
         title = "Delete keys from child namespaces",
@@ -105,24 +104,17 @@ public class PurgeKV extends Task implements RunnableTask<PurgeKV.Output> {
         boolean keyFiltering = StringUtils.isNotBlank(renderedKeyPattern);
         runContext.logger().info("purging {} namespaces: {}", kvNamespaces.size(), kvNamespaces);
         AtomicLong count = new AtomicLong();
-        PurgeBehavior renderedBehavior;
+        KvPurgeBehavior renderedBehavior;
         if (expiredOnly != null) {
             renderedBehavior = Key.builder()
                 .expiredOnly(runContext.render(expiredOnly).as(Boolean.class).orElse(true))
                 .build();
         } else {
-            renderedBehavior = runContext.render(behavior).as(PurgeBehavior.class).orElseThrow();
+            renderedBehavior = runContext.render(behavior).as(KvPurgeBehavior.class).orElseThrow();
         }
         for (String ns : kvNamespaces) {
             KVStore kvStore = runContext.namespaceKv(ns);
-            List<KVEntry> toPurge = renderedBehavior.entriesToPurge(kvStore).stream()
-                .filter(kv -> {
-                    if (keyFiltering) {
-                        return FilenameUtils.wildcardMatch(kv.key(), renderedKeyPattern);
-                    }
-                    return true;
-                })
-                .toList();
+            List<KVEntry> toPurge = filterItems(runContext, renderedBehavior.entriesToPurge(kvStore));
             count.addAndGet(kvStore.purge(toPurge));
         }
         runContext.logger().info("purged {} keys", count.get());
@@ -132,58 +124,15 @@ public class PurgeKV extends Task implements RunnableTask<PurgeKV.Output> {
             .build();
     }
 
-    @VisibleForTesting
-    protected List<String> findNamespaces(RunContext runContext) throws IllegalVariableEvaluationException {
-        String tenantId = runContext.flowInfo().tenantId();
-        String currentNamespace = runContext.flowInfo().namespace();
-        FlowRepositoryInterface flowRepositoryInterface = ((DefaultRunContext) runContext)
-            .getApplicationContext().getBean(FlowRepositoryInterface.class);
-        List<String> distinctNamespaces = flowRepositoryInterface.findDistinctNamespace(tenantId);
-        List<String> renderedNamespaces = runContext.render(namespaces).asList(String.class);
-        String renderedNamespacePattern = runContext.render(namespacePattern).as(String.class).orElse(null);
-
-        if (!ListUtils.isEmpty(renderedNamespaces) && StringUtils.isNotBlank(renderedNamespacePattern)) {
-            throw new ValidationErrorException(List.of("Properties `namespaces` and `namespacePattern` can't be used at the same time — use one or the other."));
-        }
-
-        List<String> kvNamespaces = new ArrayList<>();
-        if (StringUtils.isNotBlank(renderedNamespacePattern)) {
-            kvNamespaces.addAll(distinctNamespaces.stream()
-                .filter(ns -> FilenameUtils.wildcardMatch(ns, renderedNamespacePattern))
-                .toList());
-        } else if (!renderedNamespaces.isEmpty()) {
-            if (runContext.render(includeChildNamespaces).as(Boolean.class).orElse(true)) {
-                kvNamespaces.addAll(distinctNamespaces.stream()
-                    .filter(ns -> {
-                        for (String renderedNamespace : renderedNamespaces) {
-                            if (ns.startsWith(renderedNamespace)) {
-                                return true;
-                            }
-                        }
-                        return false;
-                    }).toList());
-            } else {
-                kvNamespaces.addAll(distinctNamespaces.stream()
-                    .filter(ns -> {
-                        for (String renderedNamespace : renderedNamespaces) {
-                            if (ns.equals(renderedNamespace)) {
-                                return true;
-                            }
-                        }
-                        return false;
-                    }).toList());
-            }
-        } else {
-            kvNamespaces.addAll(distinctNamespaces);
-        }
-
-        FlowService flowService = ((DefaultRunContext) runContext).getApplicationContext().getBean(FlowService.class);
-        for (String ns : kvNamespaces) {
-            flowService.checkAllowedNamespace(tenantId, ns, tenantId, currentNamespace);
-        }
-        return kvNamespaces;
+    @Override
+    public Property<String> filterPattern() {
+        return keyPattern;
     }
 
+    @Override
+    public String filterTargetExtractor(KVEntry item) {
+        return item.key();
+    }
 
     @Builder
     @Getter

core/src/main/java/io/kestra/plugin/core/kv/Version.java

@@ -25,7 +25,7 @@ import java.util.stream.Collectors;
 @Getter
 @NoArgsConstructor
 @KvVersionBehaviorValidation
-public class Version extends PurgeBehavior {
+public class Version extends KvPurgeBehavior {
     @NotNull
     @JsonInclude
     @Builder.Default

core/src/main/java/io/kestra/plugin/core/log/PurgeLogs.java

@@ -7,8 +7,7 @@ import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.services.FlowService;
-import io.kestra.core.services.LogService;
+import io.kestra.core.services.ExecutionLogService;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.EqualsAndHashCode;
@@ -90,15 +89,14 @@ public class PurgeLogs extends Task implements RunnableTask<PurgeLogs.Output> {
 
     @Override
     public Output run(RunContext runContext) throws Exception {
-        LogService logService = ((DefaultRunContext)runContext).getApplicationContext().getBean(LogService.class);
-        FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
+        ExecutionLogService logService = ((DefaultRunContext)runContext).getApplicationContext().getBean(ExecutionLogService.class);
 
         // validate that this namespace is authorized on the target namespace / all namespaces
         var flowInfo = runContext.flowInfo();
         if (namespace == null){
-            flowService.checkAllowedAllNamespaces(flowInfo.tenantId(), flowInfo.tenantId(), flowInfo.namespace());
+            runContext.acl().allowAllNamespaces().check();
         } else if (!flowInfo.namespace().equals(runContext.render(namespace).as(String.class).orElse(null))) {
-            flowService.checkAllowedNamespace(flowInfo.tenantId(), runContext.render(namespace).as(String.class).orElse(null), flowInfo.tenantId(), flowInfo.namespace());
+            runContext.acl().allowNamespace(runContext.render(namespace).as(String.class).orElse(null)).check();
         }
 
         var logLevelsRendered = runContext.render(this.logLevels).asList(Level.class);

core/src/main/java/io/kestra/plugin/core/namespace/DeleteFiles.java

@@ -119,7 +119,7 @@ public class DeleteFiles extends Task implements RunnableTask<Output> {
         long count = matched
             .stream()
             .map(Rethrow.throwFunction(file -> {
-                if (namespace.delete(NamespaceFile.of(renderedNamespace, Path.of(file.path().replace("\\","/"))).storagePath())) {
+                if (!namespace.delete(Path.of(file.path().replace("\\", "/"))).isEmpty()) {
                     logger.debug(String.format("Deleted %s", (file.path())));
 
                     if (Boolean.TRUE.equals(deleteParent)) {
@@ -147,13 +147,7 @@ public class DeleteFiles extends Task implements RunnableTask<Output> {
             .forEach(folderPath -> {
                 try {
                     if (namespace.isDirectoryEmpty(folderPath)) {
-                        // Create proper NamespaceFile for folder with trailing slash
-                        NamespaceFile folder = NamespaceFile.of(
-                            namespace.namespace(),
-                            URI.create(folderPath + "/")
-                        );
-
-                        if (namespace.deleteDirectory(folder)) {
+                        if (!namespace.delete(Path.of(folderPath + "/")).isEmpty()) {
                             logger.debug("Deleted empty folder: {}", folderPath);
                         }
                     }

core/src/main/java/io/kestra/plugin/core/namespace/FilesPurgeBehavior.java

@@ -0,0 +1,29 @@
+package io.kestra.plugin.core.namespace;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
+import io.kestra.core.storages.Namespace;
+import io.kestra.core.storages.NamespaceFile;
+import io.kestra.plugin.core.kv.Version;
+import io.micronaut.core.annotation.Introspected;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.io.IOException;
+import java.util.List;
+
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", visible = true, include = JsonTypeInfo.As.EXISTING_PROPERTY)
+@JsonSubTypes({
+    @JsonSubTypes.Type(value = Version.class, name = "version")
+})
+@Getter
+@NoArgsConstructor
+@SuperBuilder
+@Introspected
+public abstract class FilesPurgeBehavior {
+    abstract public String getType();
+
+    protected abstract List<NamespaceFile> entriesToPurge(String tenantId, Namespace namespaceStorage) throws IOException;
+}