Repository: kestra-io/kestra (java)

Issue:
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.





Pull request description:
feat(core): add NamespaceFiles props on WorkingDirectory
close #2405


Patch:
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 000000000..54a8d9a34
--- /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 000000000..5be714a06
--- /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 000000000..7e8d1732a
--- /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 0aef45701..9c0ac11e4 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 @@ public class Worker implements Runnable, AutoCloseable {
             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 8a5a69309..c47a77891 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.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.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 @@ import javax.validation.constraints.NotNull;
     }
 )
 @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 class WorkingDirectory extends Sequential {
             .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 class WorkingDirectory extends Sequential {
                 // 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);
         }
     }
 


Test patch:
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 000000000..f7544429b
--- /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 2de67b653..e3497c696 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.exceptions.InternalException;
 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 @@ public class WorkingDirectoryTest extends AbstractMemoryRunnerTest {
         suite.taskRun(runnerUtils);
     }
 
+    @Test
+    void namespaceFiles() throws TimeoutException, InternalException, IOException {
+        suite.namespaceFiles(runnerUtils);
+    }
+
     @Singleton
     public static class Suite {
         @Inject
@@ -108,7 +113,30 @@ public class WorkingDirectoryTest extends AbstractMemoryRunnerTest {
             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 000000000..3419b2086
--- /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 000000000..8bdeb7664
--- /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"