Repository: kestra-io/kestra (java)

Issue:
Allow error block to access {{taskrun.value}} of failed ForEach 
### Feature description

Take the example flow here. If this flow breaks during an iteration of ForEach, we should be able to access the value at the current `{{taskrun.value}}` . However currently this is not passed through as we are no longer in the flowable context

```
id: errors
namespace: demo

variables:
  current_iteration: ""

tasks:
- id: iterateObjects
  type: io.kestra.plugin.core.flow.ForEach
  concurrencyLimit: 1
  values: '["/tmp/kestra-wd/data/good.csv", "/tmp/kestra-wd/data/error.csv"]'
  tasks:
    - id: debugCurrentIteration
      type: io.kestra.plugin.core.debug.Return
      format: "{{ taskrun.value }}"

    - id: printFileName
      type: io.kestra.plugin.core.log.Log
      message: Let's read CSV {{ taskrun.value }}

    - id: downloadFile
      type: io.kestra.plugin.core.http.Download
      uri: '{{ taskrun.value }}'
  errors:
    - id: printErrorMsg
      type: io.kestra.plugin.core.log.Log
      message: Error reading {{ taskrun.value }}
```

Pull request description:
feat!: add a new method "taskWithState"
I've then implemented a new method `taskWithState` that basically help to filter easily the task list and display it in a more readable format:

![image](https://github.com/user-attachments/assets/a8d32707-8aa1-4475-a4c7-a02f7bbbc69d)

This also handle when the value is "state"

close #5653 

Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/RunVariables.java b/core/src/main/java/io/kestra/core/runners/RunVariables.java
index 0d70fd4a7..ac9ced02d 100644
--- a/core/src/main/java/io/kestra/core/runners/RunVariables.java
+++ b/core/src/main/java/io/kestra/core/runners/RunVariables.java
@@ -12,7 +12,6 @@ import io.kestra.core.models.flows.input.SecretInput;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.utils.ListUtils;
-import io.kestra.core.utils.MapUtils;
 import lombok.AllArgsConstructor;
 import lombok.With;
 
@@ -78,7 +77,7 @@ public final class RunVariables {
     static Map<String, Object> of(final FlowInterface flow) {
         ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
         builder.put("id", flow.getId())
-               .put("namespace", flow.getNamespace());
+            .put("namespace", flow.getNamespace());
 
         Optional.ofNullable(flow.getRevision())
             .ifPresent(revision ->  builder.put("revision", revision));
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/Extension.java b/core/src/main/java/io/kestra/core/runners/pebble/Extension.java
index 0be75fed1..3712f7f59 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/Extension.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/Extension.java
@@ -161,6 +161,7 @@ public class Extension extends AbstractExtension {
         functions.put("randomPort", new RandomPortFunction());
         functions.put("fileExists", fileExistsFunction);
         functions.put("isFileEmpty", isFileEmptyFunction);
+        functions.put("tasksWithState", new TasksWithState());
         return functions;
     }
 
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/TasksWithState.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/TasksWithState.java
new file mode 100644
index 000000000..46be39028
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/TasksWithState.java
@@ -0,0 +1,85 @@
+package io.kestra.core.runners.pebble.functions;
+
+import io.pebbletemplates.pebble.error.PebbleException;
+import io.pebbletemplates.pebble.extension.Function;
+import io.pebbletemplates.pebble.template.EvaluationContext;
+import io.pebbletemplates.pebble.template.EvaluationContextImpl;
+import io.pebbletemplates.pebble.template.PebbleTemplate;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TasksWithState implements Function {
+    public List<String> getArgumentNames() {
+        return List.of("state");
+    }
+
+    @Override
+    public Object execute(Map<String, Object> args, PebbleTemplate self, EvaluationContext context, int lineNumber) {
+        if (!args.containsKey("state")) {
+            throw new PebbleException(null, "The 'tasksWithState' function expects an argument 'state'.", lineNumber, self.getName());
+        }
+        String stateToFilter = ((String) args.get("state")).toUpperCase();
+
+        EvaluationContextImpl evaluationContext = (EvaluationContextImpl) context;
+
+        Map<String, Object> globalTasksMap = (Map<String, Object>) evaluationContext.getScopeChain().getGlobalScopes().stream()
+            .flatMap(scope -> scope.getKeys().stream())
+            .distinct()
+            .filter(key -> "tasks".equals(key)) // Filter for the "tasks" key specifically
+            .collect(HashMap::new, (m, k) -> m.put(k, context.getVariable(k)), HashMap::putAll)
+            .get("tasks");
+
+        List<Map<String, Object>> filteredTasks = new ArrayList<>();
+
+        if (globalTasksMap != null) {
+            globalTasksMap.forEach((taskId, taskDetailsObj) -> {
+                if (taskDetailsObj instanceof Map) {
+                    Map<String, Object> taskDetailsMap = (Map<String, Object>) taskDetailsObj;
+                    taskDetailsMap.forEach((key, valueObj) -> {
+                        Map<String, Object> transformedTask = new HashMap<>();
+                        transformedTask.put("taskId", taskId);
+
+                        if ("state".equals(key)) {
+                            if (valueObj instanceof String) {
+                                String state = ((String) valueObj).toUpperCase();
+                                if (stateToFilter.equals(state)) {
+                                    transformedTask.put("state", state);
+                                    filteredTasks.add(transformedTask);
+                                }
+
+                            }
+                            else if (valueObj instanceof Map) {
+                                Map<String, Object> nestedMap = (Map<String, Object>) valueObj;
+                                Object stateFromNested = nestedMap.get("state");
+                                if (stateFromNested instanceof String) {
+                                    String state = ((String) stateFromNested).toUpperCase();
+                                    if (stateToFilter.equals(state)) {
+                                        transformedTask.put("state", state);
+                                        transformedTask.put("value", "state");
+                                        filteredTasks.add(transformedTask);
+                                    }
+                                }
+                            }
+                        } else if (valueObj instanceof Map) {
+                            Map<String, Object> nestedMap = (Map<String, Object>) valueObj;
+                            Object stateFromNested = nestedMap.get("state");
+                            if (stateFromNested instanceof String) {
+                                String state = ((String) stateFromNested).toUpperCase();
+                                if (stateToFilter.equals(state)) {
+                                    transformedTask.put("state", state);
+                                    transformedTask.put("value", key);
+                                    filteredTasks.add(transformedTask);
+                                }
+                            }
+                        }
+                    });
+                }
+            });
+        }
+
+        return filteredTasks;
+    }
+}
\ No newline at end of file


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/RunContextTest.java b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
index 0c585b11b..e47518ab3 100644
--- a/core/src/test/java/io/kestra/core/runners/RunContextTest.java
+++ b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
@@ -63,7 +63,6 @@ import java.util.concurrent.TimeoutException;
 import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.within;
-import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @KestraTest(startRunner = true)
@@ -293,7 +292,6 @@ class RunContextTest {
         assertThat(rendered.get("key")).isEqualTo("value");
     }
 
-
     @Test
     @EnabledIfEnvironmentVariable(named = "SECRET_PASSWORD", matches = ".*")
     void secretTrigger() throws IllegalVariableEvaluationException {
@@ -339,6 +337,21 @@ class RunContextTest {
         assertThrows(ConstraintViolationException.class, () -> runContext.validate(testBean));
     }
 
+    @Test
+    @ExecuteFlow("flows/invalids/foreach-switch-failed.yaml")
+    void failedTasksVariable(Execution execution) throws Exception {
+
+        assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
+
+        TaskRun taskRun = execution.getTaskRunList().stream()
+            .filter(tr -> tr.getTaskId().equals("errorforeach"))
+            .findFirst()
+            .orElseThrow(() -> new Exception("TaskRun not found"));
+
+        assertThat(taskRun.getOutputs().get("value").toString().contains("{\"state\":\"FAILED\",\"value\":\"2\",\"taskId\":\"switch\"}")).isEqualTo(true);
+
+    }
+
     @SuperBuilder
     @ToString
     @EqualsAndHashCode
diff --git a/core/src/test/resources/flows/invalids/foreach-switch-failed.yaml b/core/src/test/resources/flows/invalids/foreach-switch-failed.yaml
new file mode 100644
index 000000000..e1ccb7111
--- /dev/null
+++ b/core/src/test/resources/flows/invalids/foreach-switch-failed.yaml
@@ -0,0 +1,23 @@
+id: foreach-switch-failed
+namespace: io.kestra.tests
+
+tasks:
+  - id: foreach
+    type: io.kestra.plugin.core.flow.ForEach
+    values: [1,2]
+    tasks:
+      - id: switch
+        type: io.kestra.plugin.core.flow.Switch
+        value: "{{ taskrun.value  }}"
+        cases:
+          1:
+            - id: log
+              type: io.kestra.plugin.core.log.Log
+              message: "hello"
+          2:
+            - id: fail
+              type: io.kestra.plugin.core.execution.Fail
+    errors:
+      - id: errorforeach
+        type: io.kestra.plugin.core.debug.Return
+        format: "{{ tasksWithState('failed') }}"
\ No newline at end of file