Repository: kestra-io/kestra
Issue: Simplify `EachParallel` with `WorkingDirectory` child tasks
Labels: bug

### Issue description

This flow works:

```yaml
id: reproducer
namespace: dev

tasks:
  - id: parallel
    type: io.kestra.core.tasks.flows.EachParallel
    value: ["1", "2", "3", "4", "5", "6"]
    tasks:
      - id: seq
        type: io.kestra.core.tasks.flows.Sequential
        tasks:
        - id: pythonScripts
          type: io.kestra.core.tasks.flows.WorkingDirectory
          tasks:
            - id: cloneRepository
              type: io.kestra.plugin.git.Clone
              url: https://github.com/kestra-io/scripts
              branch: main
            - id: python
              type: io.kestra.plugin.scripts.python.Commands
              runner: DOCKER
              docker:
                image: ghcr.io/kestra-io/pydata:latest
              commands:
                - python etl/parametrized.py {{ parents[0].taskrun.value }}
```

however, it seems that it should already work without having to add the `Sequential` task, because `WorkingDirectory` already runs everything sequentially. However, without Sequential, the `taskrun.value` cannot be accessed:

```yaml
id: reproducer
namespace: dev

tasks:
  - id: parallel
    type: io.kestra.core.tasks.flows.EachParallel
    value: ["1", "2", "3", "4", "5", "6"]
    tasks:
      - id: pythonScripts
        type: io.kestra.core.tasks.flows.WorkingDirectory
        tasks:
          - id: cloneRepository
            type: io.kestra.plugin.git.Clone
            url: https://github.com/kestra-io/scripts
            branch: main
          - id: python
            type: io.kestra.plugin.scripts.python.Commands
            runner: DOCKER
            docker:
              image: ghcr.io/kestra-io/pydata:latest
            commands:
              - python etl/parametrized.py {{ taskrun.value }}
```

this gives - `Missing variable: 'value' on 'python etl/parametrized.py {{ taskrun.value }}' at line 1`

cc @loicmathieu 

Issue comments:
As you're in a nested flowable, it should be `parent.taskRun.value` to access the Each value.
What is strange is that it seems to works with the expression eval (the property `taskRun` also works):
![image](https://github.com/kestra-io/kestra/assets/37600690/633fe61f-eb1e-4c1c-8ed6-050f72e2c339)

The fix didn't seems to solve the issue.

The following flow fail with a variable exception (parent is not present):

```yaml
id: working-dir-value
namespace: dev
tasks:
  - id: each
    type: io.kestra.core.tasks.flows.EachSequential
    value: ["1", "2", "3"]
    tasks:
      - id: workingDir
        type: io.kestra.core.tasks.flows.WorkingDirectory
        tasks:
          - id: log
            type: io.kestra.core.tasks.log.Log
            message: "{{ parent }}"
```
@loicmathieu  As stated here : https://github.com/kestra-io/kestra/pull/2293
When using a WorkingDirectory, you're not adding a flowable level (even if workingDirectory is a flowable)
So parent wont exists because there is only one level of flowable, so you need to use `taskrun.value`
I see you reopened the issue -- worth checking this follow-up issue then https://github.com/kestra-io/kestra/issues/2513 

The main request is avoiding a new vocabulary through `workerTaskrun.value` and replace it with:

1. taskrun.value
2. parent.taskrun.value
3. or if needed: worker.taskrun.value

for consistency. the 1 and 2 are preferable. 

The comment from @loicmathieu is correct, using both taskrun.value and parent.taskrun.value will fail here

![image](https://github.com/kestra-io/kestra/assets/86264395/cf540c05-73e4-425f-adbb-bc0d040e3304)
![image](https://github.com/kestra-io/kestra/assets/86264395/2ba60421-17f1-4345-b802-4dc532ced789)

the workerTaskrun.value is the only one that works atm:

```yaml
id: working-dir-value
namespace: dev
tasks:
  - id: each
    type: io.kestra.core.tasks.flows.EachSequential
    value: ["1", "2", "3"]
    tasks:
      - id: workingDir
        type: io.kestra.core.tasks.flows.WorkingDirectory
        tasks:
          - id: log
            type: io.kestra.core.tasks.log.Log
            message: "{{ workerTaskrun.value }}"
```

it would be great if using message: "{{ taskrun.value }}" would work
`taskrun.value` should works, and this is the recommended way, the same way you should be doing if there was not  workingDirectory

Again, please refer to the comment [here](https://github.com/kestra-io/kestra/pull/2293#issuecomment-1764379306) to read what have been done

edit: Works properly when there is more than 2 flowables (not counting workingDirectory)
need a small fix for when there is only 1 level
done in https://github.com/kestra-io/kestra/pull/2528 

Pull request review comments:
Can you merge those two methods in one ?

We really don't want a method that allow mutation of variables to be public!

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index 330aaeedf8..f849406dd1 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -426,6 +426,24 @@ public RunContext forWorker(ApplicationContext applicationContext, WorkerTrigger
         return forScheduler(workerTrigger.getTriggerContext(), workerTrigger.getTrigger());
     }
 
+    public RunContext forWorkerDirectory(ApplicationContext applicationContext, WorkerTask workerTask) {
+        forWorker(applicationContext, workerTask);
+
+        Map<String, Object> clone = new HashMap<>(this.variables);
+
+        clone.put("workerTaskrun", clone.get("taskrun"));
+        if (clone.containsKey("parents")) {
+            clone.put("parents", clone.get("parents"));
+        }
+        if (clone.containsKey("parent")) {
+            clone.put("parent", clone.get("parent"));
+        }
+
+        this.variables = ImmutableMap.copyOf(clone);
+
+        return this;
+    }
+
     public String render(String inline) throws IllegalVariableEvaluationException {
         return variableRenderer.render(inline, this.variables);
     }
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 dd2f5ba739..19315cab09 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -149,7 +149,7 @@ private void handleTask(WorkerTask workerTask) {
         if (workerTask.getTask() instanceof RunnableTask) {
             this.run(workerTask, true);
         } else if (workerTask.getTask() instanceof WorkingDirectory workingDirectory) {
-            RunContext runContext = workerTask.getRunContext().forWorker(applicationContext, workerTask);
+            RunContext runContext = workerTask.getRunContext().forWorkerDirectory(applicationContext, workerTask);
 
             try {
                 workingDirectory.preExecuteTasks(runContext, workerTask.getTaskRun());
@@ -158,7 +158,6 @@ private void handleTask(WorkerTask workerTask) {
                     if (Boolean.TRUE.equals(currentTask.getDisabled())) {
                         continue;
                     }
-
                     WorkerTask currentWorkerTask = workingDirectory.workerTask(
                         workerTask.getTaskRun(),
                         currentTask,


Test patch:
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 6aca4e1209..e2e00ac344 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
@@ -4,6 +4,7 @@
 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;
@@ -43,6 +44,11 @@ void cache() throws TimeoutException, IOException {
         suite.cache(runnerUtils);
     }
 
+    @Test
+    void taskrun() throws TimeoutException {
+        suite.taskRun(runnerUtils);
+    }
+
     @Singleton
     public static class Suite {
         @Inject
@@ -93,5 +99,14 @@ public void cache(RunnerUtils runnerUtils) throws TimeoutException, IOException
             assertThat(execution.getTaskRunList(), hasSize(2));
             assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
         }
+
+        public void taskRun(RunnerUtils runnerUtils) throws TimeoutException {
+            Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "working-directory-taskrun");
+
+            assertThat(execution.getTaskRunList(), hasSize(6));
+            assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+            assertThat(((String) execution.getTaskRunList().get(5).getOutputs().get("value")), containsString("{\"taskrun\":{\"value\":\"1\"}}"));
+
+        }
     }
 }
diff --git a/core/src/test/resources/flows/valids/working-directory-taskrun.yml b/core/src/test/resources/flows/valids/working-directory-taskrun.yml
new file mode 100644
index 0000000000..8a1899dd91
--- /dev/null
+++ b/core/src/test/resources/flows/valids/working-directory-taskrun.yml
@@ -0,0 +1,23 @@
+id: working-directory-taskrun
+namespace: io.kestra.tests
+
+tasks:
+  - id: parallel
+    type: io.kestra.core.tasks.flows.EachParallel
+    value: ["1"]
+    tasks:
+      - id: seq
+        type: io.kestra.core.tasks.flows.Sequential
+        tasks:
+          - id: workingDir
+            type: io.kestra.core.tasks.flows.WorkingDirectory
+            tasks:
+              - id: log-taskrun
+                type: io.kestra.core.tasks.debugs.Return
+                format: "{{ workerTaskrun }}"
+              - id: log-workerparents
+                type: io.kestra.core.tasks.debugs.Return
+                format:  "{{ parents }}"
+              - id: log-workerparent
+                type: io.kestra.core.tasks.debugs.Return
+                format:  "{{ parent }}"
\ No newline at end of file