```suggestion
    private List<String> include;
```
given that it's a list, it seems better to use include as in "Include all those listed paths" ```suggestion
    private List<String> exclude;
```
same, given that it's a list, it seems better to use exclude as in "Exclude all those listed paths" Wouldn't it be easier to use the `NamespaceFiles` object here and have a way to inject everything from it?
Like a `all` boolean or just document that to inject everything one should configure `include: **/*`?

I wonder if those multi-types that we often use are easy to use for not too technical people. Wouldn't it be easier to use the `NamespaceFiles` object here and have a way to inject everything from it?
Like a `all` boolean or just document that to inject everything one should configure `include: **/*`?

I wonder if those multi-types that we often use are easy to use for not too technical people. Cache was designed to be fail-tolerant, if you didn't succeed in caching an error is displayed but the task is still runs. You change the behaviour here and I'm not sure it's a good idea as the task can still run even if it fails to use its cache. I don't thin it's a good idea, in Kestra, parallelism is at the worker thread level not at the task level.
Moreover, parallel stream uses the default forkJoinPool so it may be less performant than a standard stream if multiple tasks are executed in parallel. ```suggestion
    public List<URI> list(RunContext runContext, String tenantId, String namespace, Path basePath, Object namespaceFiles) throws Exception {
```

It didn't really inject the files but list the files to be injected by the task that uses it so the name is kind of misleading. We discussed it initially via Slack and the initial proposal was to have 2 properties: one boolean `namespaceFiles` whether to fetch all namespace files and then if true, optional `namespaceFilesFilter` property with include/exclude. Given that there is some risk of confusion of using the filter without setting the boolean to true (it must be set to false by default), we went for that option with 1 property

good point though and well spotted for now, a good tradeoff might be if we add maybe 2-3 examples in the task docs to show the usage Better to add them in the namespace file documentation as we will use this in multiple tasks and we will not add those 2-3 examples in all tasks (the script tasks already have a lot of examples).
We may add a link to the documentation for that. Cache should never failed, I don't see a reason to let failed a cache failed silently and even more that the error will be on the worker and not on the user side (ui)  this method copy the files on the directory and return the list of copied files, inject is better than list ```suggestion
        title = "Whether to enable namespace files to be loaded into the working directory"
``` ```suggestion
        title = "A list of filters to include only matching glob patterns"
```
could be more explicit this way could be also: "Filters to include only matching glob patterns" if you don't want to include the info that it's a list 👍  ```suggestion
        title = "A list of filters to exclude matching glob patterns"
``` ```suggestion
        description = "Inject namespace files to this task. When enabled, it will, by default, load all namespace files into the working directory. However, you can use the `include` or `exclude` properties to limit which namespace files will be injected."
```
Inject namespace files into the Execution context so that they are accessible from the execution's working directory
## The `namespaceFiles` property

We want to add an **optional** `namespaceFiles` property, which can be set either to:
- a `boolean` — by default, this property is set to `false`; if set to `true`, we inject/load all Namespace Files from that specific namespace into the working directory of that task
- or a `map` with `include` and `exclude` keys and values being a list of strings. Those strings are regular expressions allowing to include or exclude specific files or directories.

## `WorkingDirectory` as a first task to include `namespaceFiles` property

First, we want to add that property to the `WorkingDirectory` task. Then, we can add the same property to other tasks in plugins listed in the section below. Example with the `include` pattern to only load namespace files from the `scripts` directory:

```yaml
id: include_scripts
namespace: dev

tasks:
  - id: wdir
    type: io.kestra.core.tasks.flows.WorkingDirectory
    namespaceFiles:
      include:
        - "scripts/**"
    tasks:
      - id: myscript
        type: io.kestra.plugin.scripts.python.Commands
        commands:
          - python scripts/myscript.py
```

Simple example loading all namespace files into the task's working directory:

```yaml
id: include_all_namespace_files
namespace: dev

tasks:
  - id: wdir
    type: io.kestra.core.tasks.flows.WorkingDirectory
    namespaceFiles: true
    tasks:
      - id: myscript
        type: io.kestra.plugin.scripts.python.Commands
        commands:
          - python scripts/myscript.py
```




## Problem to be solved 

We want to make it easy to orchestrate entire projects containing:
- dbt core models and tests
- SqlMesh
- custom SQL queries
- custom Python scripts and entire modules
- Terraform modules
- Ansible projects
- Docker image builds
- CloudQuery ingestion sync files
- ...and many more.

Currently, users would need to store those in a Git repository and use a combination of `WorkingDirectory` and a `git.Clone` task.

To make that easier, we want to allow them to store all these projects as Namespace Files tied to a given namespace. Those files can be used as if you'd be working with a local directory. This way, it's possible to use as simple syntax as follows to orchestrate a Python script stored as a Namespace File in the directory `scripts`:

```yaml
id: users
namespace: dev

tasks:
  - id: getUsers
    type: io.kestra.plugin.scripts.python.Commands
    docker:
      image: ghcr.io/kestra-io/pydata:latest
    commands:
      - python scripts/get_users.py
    namespaceFiles:
      include:
        - "scripts/**" 
```

![image](https://github.com/kestra-io/kestra/assets/86264395/22ab7c2f-7380-4b25-bb8e-073a1d6abc54)

## Task types that will need the `namespaceFiles` property

Inject the namespace files only for the following tasks, as only those can benefit from them. The syntax may be implemented as described in this issue but on a task level https://github.com/kestra-io/kestra/issues/2341 


```yaml
id: myflow
namespace: dev

tasks:
  - id: dbt-build
    type: io.kestra.plugin.dbt.cli.DbtCLI
    runner: DOCKER
    docker:
      image: ghcr.io/kestra-io/dbt-duckdb
    commands:
      - dbt build
    namespaceFiles:
      include:
        - "dbt/**"  # fetch all files from internal storage from that repository
        - dbt_project.yml
        - profiles.yml
        - packages.yml
      exclude:
        - "dbt/*.py"
```

- `WorkingDirectory`

- https://kestra.io/plugins/plugin-script-python/tasks/io.kestra.plugin.scripts.python.commands
- https://kestra.io/plugins/plugin-script-r/tasks/io.kestra.plugin.scripts.r.commands
- https://kestra.io/plugins/plugin-script-julia/tasks/io.kestra.plugin.scripts.julia.commands
- https://kestra.io/plugins/plugin-script-shell/tasks/io.kestra.plugin.scripts.shell.commands
- https://kestra.io/plugins/plugin-script-powershell/tasks/io.kestra.plugin.scripts.powershell.commands
- https://kestra.io/plugins/plugin-script-node/tasks/io.kestra.plugin.scripts.node.commands
- https://kestra.io/plugins/plugin-malloy/tasks/io.kestra.plugin.malloy.cli
- https://kestra.io/plugins/plugin-aws/tasks/cli/io.kestra.plugin.aws.cli.awscli
- https://kestra.io/plugins/plugin-azure/tasks/cli/io.kestra.plugin.azure.cli.azcli
- https://kestra.io/plugins/plugin-gcp/tasks/cli/io.kestra.plugin.gcp.cli.gcloudcli
- https://kestra.io/plugins/plugin-dbt/tasks/cli/io.kestra.plugin.dbt.cli.dbtcli
- https://kestra.io/plugins/plugin-docker/tasks/io.kestra.plugin.docker.build
- TerraformCLI
- AnsibleCLI
- ModalCLI
- SQLMesh
- Dataform
- https://kestra.io/plugins/plugin-cloudquery/tasks/io.kestra.plugin.cloudquery.cloudquerycli
- https://kestra.io/plugins/plugin-fs/tasks/ssh/io.kestra.plugin.fs.ssh.command

### Where namespace files could be considered in the future, but not for now

Unsure about the usefulness of namespace files in compression tasks, but those compression + encryption tasks could potentially take a path to namespace files: 
- https://kestra.io/plugins/plugin-compress/tasks/io.kestra.plugin.compress.archivecompress
- https://kestra.io/plugins/plugin-compress/tasks/io.kestra.plugin.compress.filecompress
- https://kestra.io/plugins/plugin-crypto/tasks/openpgp/io.kestra.plugin.crypto.openpgp.encrypt

All upload tasks could, in theory, take a file from namespace files, but also here, I am not sure how useful that would be, as namespace files are meant for code, not data files:
- https://kestra.io/plugins/plugin-databricks/tasks/dbfs/io.kestra.plugin.databricks.dbfs.upload
- AWS S3 Upload
- Azure Blob Storage Upload
- GCP Storage Upload
- BigQuery Load
- BigQuery Upload

Singer doesn't need namespace files.
Soda also doesn't.




core/src/main/java/io/kestra/core/runners/Worker.java core/src/main/java/io/kestra/core/tasks/flows/WorkingDirectory.java
core/src/main/java/io/kestra/core/models/tasks/NamespaceFiles.java core/src/main/java/io/kestra/core/models/tasks/NamespaceFilesInterface.java core/src/main/java/io/kestra/core/runners/NamespaceFilesService.java core/src/main/java/io/kestra/core/runners/Worker.java core/src/main/java/io/kestra/core/tasks/flows/WorkingDirectory.java
core/src/test/java/io/kestra/core/runners/NamespaceFilesServiceTest.java core/src/test/java/io/kestra/core/tasks/flows/WorkingDirectoryTest.java core/src/test/java/io/kestra/core/tasks/test/Read.java core/src/test/resources/flows/valids/working-directory-namespace-files.yaml
diff --git a/core/src/main/java/io/kestra/core/models/tasks/NamespaceFiles.java b/core/src/main/java/io/kestra/core/models/tasks/NamespaceFiles.java
new file mode 100644
index 0000000000..54a8d9a348
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/tasks/NamespaceFiles.java
@@ -0,0 +1,40 @@
+package io.kestra.core.models.tasks;
+
+import io.kestra.core.models.annotations.PluginProperty;
+import io.micronaut.core.annotation.Introspected;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+import javax.validation.Valid;
+
+@Builder
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@Introspected
+public class NamespaceFiles {
+    @Schema(
+        title = "Whether to enable namespace files to be loaded into the working directory"
+    )
+    @PluginProperty
+    @Builder.Default
+    private Boolean enabled = true;
+
+    @Schema(
+        title = "A list of filters to include only matching glob patterns"
+    )
+    @PluginProperty
+    @Valid
+    private List<String> include;
+
+    @Schema(
+        title = "A list of filters to exclude matching glob patterns"
+    )
+    @PluginProperty
+    @Valid
+    private List<String> exclude;
+}
diff --git a/core/src/main/java/io/kestra/core/models/tasks/NamespaceFilesInterface.java b/core/src/main/java/io/kestra/core/models/tasks/NamespaceFilesInterface.java
new file mode 100644
index 0000000000..5be714a069
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/tasks/NamespaceFilesInterface.java
@@ -0,0 +1,13 @@
+package io.kestra.core.models.tasks;
+
+import io.kestra.core.models.annotations.PluginProperty;
+import io.swagger.v3.oas.annotations.media.Schema;
+
+public interface NamespaceFilesInterface {
+    @Schema(
+        title = "Inject namespace files",
+        description = "Inject namespace files to this task. When enabled, it will, by default, load all namespace files into the working directory. However, you can use the `include` or `exclude` properties to limit which namespace files will be injected."
+    )
+    @PluginProperty
+    NamespaceFiles getNamespaceFiles();
+}
diff --git a/core/src/main/java/io/kestra/core/runners/NamespaceFilesService.java b/core/src/main/java/io/kestra/core/runners/NamespaceFilesService.java
new file mode 100644
index 0000000000..7e8d1732ad
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/NamespaceFilesService.java
@@ -0,0 +1,121 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.tasks.NamespaceFiles;
+import io.kestra.core.storages.FileAttributes;
+import io.kestra.core.storages.StorageInterface;
+import io.micronaut.core.annotation.Nullable;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static io.kestra.core.utils.Rethrow.*;
+
+@Singleton
+@Slf4j
+public class NamespaceFilesService {
+    @Inject
+    private StorageInterface storageInterface;
+
+    public List<URI> inject(RunContext runContext, String tenantId, String namespace, Path basePath, NamespaceFiles namespaceFiles) throws Exception {
+        if (!namespaceFiles.getEnabled()) {
+            return Collections.emptyList();
+        }
+
+        List<URI> list = new ArrayList<>();
+        list.addAll(recursiveList(tenantId, namespace, null));
+
+
+        list = list
+            .stream()
+            .filter(throwPredicate(f -> {
+                var file = f.getPath();
+
+                if (namespaceFiles.getExclude() != null) {
+                    boolean b = match(runContext.render(namespaceFiles.getExclude()), file);
+
+                    if (b) {
+                        return false;
+                    }
+                }
+
+                if (namespaceFiles.getInclude() != null) {
+                    boolean b = match(namespaceFiles.getInclude(), file);
+
+                    if (!b) {
+                        return false;
+                    }
+                }
+
+                return true;
+            }))
+            .collect(Collectors.toList());
+
+        copy(tenantId, namespace, basePath, list);
+
+        return list;
+    }
+
+    private URI uri(String namespace, @Nullable URI path) {
+        return URI.create(storageInterface.namespaceFilePrefix(namespace) + Optional.ofNullable(path)
+            .map(URI::getPath)
+            .orElse("")
+        );
+    }
+
+    private List<URI> recursiveList(String tenantId, String namespace, @Nullable URI path) throws IOException {
+        URI uri = uri(namespace, path);
+
+        List<URI> result = new ArrayList<>();
+        List<FileAttributes> list = storageInterface.list(tenantId, uri);
+
+        for (var file: list) {
+            URI current = URI.create((path != null ? path.getPath() : "") +  "/" + file.getFileName());
+
+            if (file.getType() == FileAttributes.FileType.Directory) {
+                result.addAll(this.recursiveList(tenantId, namespace, current));
+            } else {
+                result.add(current);
+            }
+        }
+
+        return result;
+    }
+
+    private static boolean match(List<String> patterns, String file) {
+        return patterns
+            .stream()
+            .anyMatch(s -> FileSystems
+                .getDefault()
+                .getPathMatcher("glob:" + s)
+                .matches(Paths.get(file))
+            );
+    }
+
+    private void copy(String tenantId, String namespace, Path basePath, List<URI> files) throws IOException {
+        files
+            .forEach(throwConsumer(f -> {
+                InputStream inputStream = storageInterface.get(tenantId, uri(namespace, f));
+                Path destination = Paths.get(basePath.toString(), f.getPath());
+
+                if (!destination.getParent().toFile().exists()) {
+                    //noinspection ResultOfMethodCallIgnored
+                    destination.getParent().toFile().mkdirs();
+                }
+
+                Files.copy(inputStream, destination);
+            }));
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 0aef457017..9c0ac11e49 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -152,8 +152,19 @@ private void handleTask(WorkerTask workerTask) {
             RunContext runContext = workerTask.getRunContext().forWorkerDirectory(applicationContext, workerTask);
 
             try {
-                workingDirectory.preExecuteTasks(runContext, workerTask.getTaskRun());
+                // preExecuteTasks
+                try {
+                    workingDirectory.preExecuteTasks(runContext, workerTask.getTaskRun());
+                } catch (Exception e) {
+                    runContext.logger().error("Failed preExecuteTasks on WorkingDirectory: {}", e.getMessage(), e);
+                    workerTask = workerTask.withTaskRun(workerTask.getTaskRun().withState(State.Type.FAILED));
+                    this.workerTaskResultQueue.emit(new WorkerTaskResult(workerTask));
+                    this.logTerminated(workerTask);
+
+                    return;
+                }
 
+                // execute all tasks
                 for (Task currentTask : workingDirectory.getTasks()) {
                     if (Boolean.TRUE.equals(currentTask.getDisabled())) {
                         continue;
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/WorkingDirectory.java b/core/src/main/java/io/kestra/core/tasks/flows/WorkingDirectory.java
index 8a5a69309f..c47a77891d 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/WorkingDirectory.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/WorkingDirectory.java
@@ -8,8 +8,11 @@
 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.tasks.NamespaceFiles;
+import io.kestra.core.models.tasks.NamespaceFilesInterface;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
+import io.kestra.core.runners.NamespaceFilesService;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.WorkerTask;
 import io.kestra.core.utils.IdUtils;
@@ -186,7 +189,7 @@ with open('output.json', 'w') as output_file:
     }
 )
 @WorkingDirectoryTaskValidation
-public class WorkingDirectory extends Sequential {
+public class WorkingDirectory extends Sequential implements NamespaceFilesInterface {
 
     @Schema(
         title = "Cache configuration",
@@ -198,6 +201,8 @@ public class WorkingDirectory extends Sequential {
     @PluginProperty
     private Cache cache;
 
+    private NamespaceFiles namespaceFiles;
+
     @Getter(AccessLevel.PRIVATE)
     private transient long cacheDownloadedTime = 0L;
 
@@ -230,12 +235,8 @@ public WorkerTask workerTask(TaskRun parent, Task task, RunContext runContext) {
             .build();
     }
 
-    public void preExecuteTasks(RunContext runContext, TaskRun taskRun) {
-        if (cache == null) {
-            return;
-        }
-
-        try {
+    public void preExecuteTasks(RunContext runContext, TaskRun taskRun) throws Exception {
+        if (cache != null) {
             // first, check if we need to delete the file
             if (cache.ttl != null) {
                 var maybeLastModifiedTime = runContext.getTaskCacheFileLastModifiedTime(taskRun.getNamespace(), taskRun.getFlowId(), this.getId(), taskRun.getValue());
@@ -271,8 +272,11 @@ public void preExecuteTasks(RunContext runContext, TaskRun taskRun) {
                 // Set the cacheDownloadedTime so that we can check if files has been updated later
                 cacheDownloadedTime = System.currentTimeMillis();
             }
-        } catch (IOException e) {
-            runContext.logger().error("Unable to execute WorkingDirectory pre actions", e);
+        }
+
+        if (this.namespaceFiles != null ) {
+            NamespaceFilesService namespaceFilesService = runContext.getApplicationContext().getBean(NamespaceFilesService.class);
+            namespaceFilesService.inject(runContext, taskRun.getTenantId(), taskRun.getNamespace(), runContext.tempDir(), this.namespaceFiles);
         }
     }
 

diff --git a/core/src/test/java/io/kestra/core/runners/NamespaceFilesServiceTest.java b/core/src/test/java/io/kestra/core/runners/NamespaceFilesServiceTest.java
new file mode 100644
index 0000000000..f7544429b6
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/NamespaceFilesServiceTest.java
@@ -0,0 +1,135 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.tasks.NamespaceFiles;
+import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.IdUtils;
+import io.micronaut.core.annotation.Nullable;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+
+import static io.kestra.core.utils.Rethrow.throwFunction;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+
+@MicronautTest
+class NamespaceFilesServiceTest {
+    @Inject
+    StorageInterface storageInterface;
+
+    @Inject
+    NamespaceFilesService namespaceFilesService;
+
+    @Inject
+    RunContextFactory runContextFactory;
+
+    @Test
+    public void noFilter() throws Exception {
+        Path basePath = Files.createTempDirectory("unit");
+        String namespace = "io.kestra." + IdUtils.create();
+
+        put(null, namespace, "/a/b/c/1.sql", "1");
+        put(null, namespace, "/a/1.sql", "2");
+        put(null, namespace, "/b/c/d/1.sql", "3");
+
+        List<URI> injected = namespaceFilesService.inject(
+            runContextFactory.of(),
+            null,
+            namespace,
+            basePath,
+            NamespaceFiles
+                .builder()
+                .enabled(true)
+                .build()
+        );
+
+        assertThat(injected.size(), is(3));
+
+        List<Path> tempDir = Files.walk(basePath).filter(path -> path.toFile().isFile()).toList();
+        assertThat(tempDir.size(), is(3));
+    }
+
+    @Test
+    public void filter() throws Exception {
+        Path basePath = Files.createTempDirectory("unit");
+        String namespace = "io.kestra." + IdUtils.create();
+
+        put(null, namespace, "/a/b/c/1.sql", "1");
+        put(null, namespace, "/a/3.sql", "2");
+        put(null, namespace, "/b/c/d/1.sql", "3");
+
+        List<URI> injected = namespaceFilesService.inject(
+            runContextFactory.of(),
+            null,
+            namespace,
+            basePath,
+            NamespaceFiles.builder()
+                .include(List.of("/a/**"))
+                .exclude(List.of("**/3.sql"))
+                .build()
+        );
+
+        assertThat(injected.size(), is(1));
+        assertThat(injected.get(0).getPath(), containsString("c/1.sql"));
+        List<Path> tempDir = Files.walk(basePath).filter(path -> path.toFile().isFile()).toList();
+        assertThat(tempDir.size(), is(1));
+        assertThat(tempDir.get(0).toString(), is(Paths.get(basePath.toString(), "/a/b/c/1.sql").toString()));
+    }
+
+    @Test
+    public void tenant() throws Exception {
+        String namespace = "io.kestra." + IdUtils.create();
+
+        put("tenant1", namespace, "/a/b/c/1.sql", "1");
+        put("tenant2", namespace, "/a/b/c/1.sql", "2");
+
+        RunContext runContext = runContextFactory.of();
+        List<URI> injected = namespaceFilesService.inject(
+            runContextFactory.of(),
+            "tenant1",
+            namespace,
+            runContext.tempDir(),
+            NamespaceFiles
+                .builder()
+                .enabled(true)
+                .build()
+        );
+        assertThat(injected.size(), is(1));
+
+        String content = Files.walk(runContext.tempDir()).filter(path -> path.toFile().isFile()).findFirst().map(throwFunction(Files::readString)).orElseThrow();
+        assertThat(content, is("1"));
+
+        runContext = runContextFactory.of();
+        injected = namespaceFilesService.inject(
+            runContextFactory.of(),
+            "tenant2",
+            namespace,
+            runContext.tempDir(),
+            NamespaceFiles
+                .builder()
+                .enabled(true)
+                .build()
+        );
+        assertThat(injected.size(), is(1));
+
+        content = Files.walk(runContext.tempDir()).filter(path -> path.toFile().isFile()).findFirst().map(throwFunction(Files::readString)).orElseThrow();
+        assertThat(content, is("2"));
+    }
+
+    private void put(@Nullable String tenantId, String namespace, String path, String content) throws IOException {
+        storageInterface.put(
+            tenantId,
+            URI.create(storageInterface.namespaceFilePrefix(namespace)  + path),
+            new ByteArrayInputStream(content.getBytes())
+        );
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/WorkingDirectoryTest.java b/core/src/test/java/io/kestra/core/tasks/flows/WorkingDirectoryTest.java
index 2de67b6530..e3497c6963 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/WorkingDirectoryTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/WorkingDirectoryTest.java
@@ -5,13 +5,13 @@
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
-import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.runners.RunnerUtils;
 import io.kestra.core.storages.StorageInterface;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.junit.jupiter.api.Test;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
@@ -51,6 +51,11 @@ void taskrun() throws TimeoutException, InternalException {
         suite.taskRun(runnerUtils);
     }
 
+    @Test
+    void namespaceFiles() throws TimeoutException, InternalException, IOException {
+        suite.namespaceFiles(runnerUtils);
+    }
+
     @Singleton
     public static class Suite {
         @Inject
@@ -108,7 +113,30 @@ public void taskRun(RunnerUtils runnerUtils) throws TimeoutException, InternalEx
             assertThat(execution.getTaskRunList(), hasSize(6));
             assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
             assertThat(((String) execution.findTaskRunByTaskIdAndValue("log-workerparent", List.of("1")).getOutputs().get("value")), containsString("{\"taskrun\":{\"value\":\"1\"}}"));
+        }
+
+        public void namespaceFiles(RunnerUtils runnerUtils) throws TimeoutException, InternalException, IOException {
+            put("/test/a/b/c/1.txt", "first");
+            put("/a/b/c/2.txt", "second");
+            put("/a/b/3.txt", "third");
+            put("/ignore/4.txt", "4th");
 
+            Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "working-directory-namespace-files");
+
+            assertThat(execution.getTaskRunList(), hasSize(6));
+            assertThat(execution.getState().getCurrent(), is(State.Type.WARNING));
+            assertThat(execution.findTaskRunsByTaskId("t4").get(0).getState().getCurrent(), is(State.Type.FAILED));
+            assertThat(execution.findTaskRunsByTaskId("t1").get(0).getOutputs().get("value"), is("first"));
+            assertThat(execution.findTaskRunsByTaskId("t2").get(0).getOutputs().get("value"), is("second"));
+            assertThat(execution.findTaskRunsByTaskId("t3").get(0).getOutputs().get("value"), is("third"));
+        }
+
+        private void put(String path, String content) throws IOException {
+            storageInterface.put(
+                null,
+                URI.create(storageInterface.namespaceFilePrefix("io.kestra.tests")  + path),
+                new ByteArrayInputStream(content.getBytes())
+            );
         }
     }
 }
diff --git a/core/src/test/java/io/kestra/core/tasks/test/Read.java b/core/src/test/java/io/kestra/core/tasks/test/Read.java
new file mode 100644
index 0000000000..3419b20869
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/test/Read.java
@@ -0,0 +1,40 @@
+package io.kestra.core.tasks.test;
+
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.tasks.RunnableTask;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.runners.RunContext;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import javax.validation.constraints.NotNull;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+public class Read extends Task implements RunnableTask<Read.Output> {
+    @PluginProperty
+    @NotNull
+    private String path;
+
+    @Override
+    public Read.Output run(RunContext runContext) throws Exception {
+        return Output.builder()
+            .value(Files.readString(Paths.get(runContext.tempDir().toString(), runContext.render(path))))
+            .build();
+    }
+
+    @Builder
+    @Getter
+    public static class Output implements io.kestra.core.models.tasks.Output {
+        @Schema(
+            title = "The file contents"
+        )
+        private String value;
+    }
+}
diff --git a/core/src/test/resources/flows/valids/working-directory-namespace-files.yaml b/core/src/test/resources/flows/valids/working-directory-namespace-files.yaml
new file mode 100644
index 0000000000..8bdeb76640
--- /dev/null
+++ b/core/src/test/resources/flows/valids/working-directory-namespace-files.yaml
@@ -0,0 +1,26 @@
+id: working-directory-namespace-files
+namespace: io.kestra.tests
+
+tasks:
+  - id: allow
+    type: io.kestra.core.tasks.flows.AllowFailure
+    tasks:
+      - id: worker
+        type: io.kestra.core.tasks.flows.WorkingDirectory
+        namespaceFiles:
+          enabled: true
+          exclude:
+            - /ignore/**
+        tasks:
+          - id: t1
+            type: io.kestra.core.tasks.test.Read
+            path: "/test/a/b/c/1.txt"
+          - id: t2
+            type: io.kestra.core.tasks.test.Read
+            path: "/a/b/c/2.txt"
+          - id: t3
+            type: io.kestra.core.tasks.test.Read
+            path: "/a/b/3.txt"
+          - id: t4
+            type: io.kestra.core.tasks.test.Read
+            path: "/ignore/4.txt"
