feat(inputs): FORM input type with a Next/Back execution wizard (#16655)

Add the FORM input type and render FORM-grouped inputs as a multi-step Next/Back
wizard (shared by the Flow Execute modal and EE Apps), and surface input
render/resolution failures eagerly.

- Wizard: one step per FORM + one per contiguous run of ungrouped inputs + a
  recap; per-section Edit, Back/Next, value persistence; delay-gated "Loading"
  on Next; payload-signature dedup + in-flight coalesce of validates.
- unflattenToForms() rebuilds the FORM tree from flat dotted leaves + formGroups
  so EE Apps drives the same wizard; promisified @Validation emit.
- Surface render errors (a SELECT expression/subflow(), or an input defaults
  Pebble expression, that fails to render) as soon as the input is shown — via a
  renderError flag on InputOutputValidationException, serialized in the
  execute-form validate response and read by InputsForm.inputError. Plain value
  errors (e.g. required-but-empty) stay gated until interaction. updateDefaults
  no longer pre-fills an unrendered expression-default (which masked the error).
- FORM child labels show the bare name; unit + storybook coverage.

Author: brian-mulier-p

core/src/main/java/io/kestra/core/exceptions/InputOutputValidationException.java

@@ -12,15 +12,37 @@
 public class InputOutputValidationException extends KestraRuntimeException {
     private static final long serialVersionUID = 1L;
 
+    /**
+     * Whether this error comes from rendering/resolving the input (e.g. a SELECT's {@code expression}
+     * or an input's {@code defaults} using a Pebble function that failed) rather than from validating a
+     * provided value (e.g. a required input left empty). A render error means the field itself is broken.
+     */
+    private final boolean renderError;
+
     public InputOutputValidationException(String message) {
+        this(message, false);
+    }
+
+    public InputOutputValidationException(String message, boolean renderError) {
         super(message);
+        this.renderError = renderError;
+    }
+
+    public boolean isRenderError() {
+        return renderError;
     }
 
     public static InputOutputValidationException of(String message, Input<?> input) {
         String inputMessage = "Invalid value for input" + " `" + input.getId() + "`. Cause: " + message;
         return new InputOutputValidationException(inputMessage);
     }
 
+    /** As {@link #of(String, Input)}, but flags the error as a render/resolution failure (broken field). */
+    public static InputOutputValidationException ofRenderError(String message, Input<?> input) {
+        String inputMessage = "Invalid value for input" + " `" + input.getId() + "`. Cause: " + message;
+        return new InputOutputValidationException(inputMessage, true);
+    }
+
     public static InputOutputValidationException of(String message, Output output) {
         String outputMessage = "Invalid value for output" + " `" + output.getId() + "`. Cause: " + message;
         return new InputOutputValidationException(outputMessage);

core/src/main/java/io/kestra/core/models/flows/FlowInterface.java

@@ -41,8 +41,32 @@ public interface FlowInterface extends FlowId, SoftDeletable<FlowInterface>, Ten
 
     List<Label> getLabels();
 
+    /**
+     * The inputs as authored in the flow source, which may contain {@link io.kestra.core.models.flows.input.FormInput}
+     * grouping nodes. This is the structure meant for serialization and the UI.
+     * <p>
+     * <strong>Do not use this when resolving, validating, or turning inputs into variables</strong> — use
+     * {@link #resolvableInputs()} instead, which expands every {@code FORM} into its dotted-path leaves. Reaching for
+     * {@code getInputs()} on a resolution path and forgetting to expand silently leaks {@code FORM} nodes into the
+     * flat keyspace.
+     */
     List<Input<?>> getInputs();
 
+    /**
+     * The inputs flattened for resolution: every {@link io.kestra.core.models.flows.input.FormInput} is expanded into
+     * its dotted-path leaves (see {@link Input#expandToLeaves(List)}).
+     * <p>
+     * Use this anywhere inputs are resolved, validated, or turned into variables. {@link #getInputs()} returns the
+     * authored structure (which may contain {@code FORM} nodes) and is meant for serialization and the UI only —
+     * reaching for it during resolution and forgetting to expand silently leaks {@code FORM} nodes into the flat
+     * keyspace.
+     */
+    @JsonIgnore
+    default List<Input<?>> resolvableInputs() {
+        List<Input<?>> inputs = getInputs();
+        return inputs == null ? List.of() : Input.expandToLeaves(inputs);
+    }
+
     List<Output> getOutputs();
 
     Map<String, Object> getVariables();

core/src/main/java/io/kestra/core/models/flows/Input.java

@@ -5,6 +5,7 @@
 
 import io.kestra.core.models.flows.input.*;
 import io.kestra.core.models.property.Property;
+import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.validations.InputValidation;
 
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -18,6 +19,10 @@
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
 @SuperBuilder
 @Getter
 @NoArgsConstructor
@@ -41,6 +46,7 @@
         @JsonSubTypes.Type(value = MultiselectInput.class, name = "MULTISELECT"),
         @JsonSubTypes.Type(value = YamlInput.class, name = "YAML"),
         @JsonSubTypes.Type(value = EmailInput.class, name = "EMAIL"),
+        @JsonSubTypes.Type(value = FormInput.class, name = "FORM"),
     }
 )
 @InputValidation
@@ -90,4 +96,59 @@ public abstract class Input<T> implements Data {
     String displayName;
 
     public abstract void validate(T input) throws ConstraintViolationException;
+
+    /**
+     * Expands every {@link io.kestra.core.models.flows.input.FormInput} into copies of its children whose id is
+     * rewritten to the dotted path ({@code environment} + {@code .} + {@code region} -> {@code environment.region}).
+     * <p>
+     * The resolution core keys inputs by {@link #getId()} and reassembles dotted keys into a nested map via
+     * {@link io.kestra.core.utils.MapUtils#flattenToNestedMap(java.util.Map)}, so a leaf carrying the dotted id is all
+     * that is needed for the nested payload to materialize. {@code dependsOn} is intentionally NOT rewritten — authors
+     * reference siblings by their full dotted path; an unresolved bare ref is rejected loudly by flow validation.
+     * Forms cannot be nested (enforced by validation), so expansion is single-level.
+     *
+     * @return a flat list of leaf inputs with no {@code FORM} node surviving, or the input list unchanged when null/empty.
+     */
+    public static List<Input<?>> expandToLeaves(List<Input<?>> inputs) {
+        if (inputs == null || inputs.isEmpty()) {
+            return inputs;
+        }
+
+        List<Input<?>> leaves = new ArrayList<>();
+        for (Input<?> input : inputs) {
+            if (input instanceof io.kestra.core.models.flows.input.FormInput form) {
+                if (form.getInputs() != null) {
+                    for (Input<?> child : form.getInputs()) {
+                        leaves.add(copyWithId(child, form.getId() + "." + child.getId()));
+                    }
+                }
+            } else {
+                leaves.add(input);
+            }
+        }
+        return leaves;
+    }
+
+    /**
+     * Copies {@code input} with its {@code id} replaced by {@code newId}, preserving the concrete subtype.
+     * <p>
+     * Done via a Jackson round-trip rather than a builder: {@code @SuperBuilder(toBuilder=...)} does not generate
+     * {@code toBuilder()} on this abstract class, and the resolution core casts leaves to their concrete subtype
+     * ({@code (ArrayInput) input}, {@code case StringInput i -> ...}). The round-trip re-resolves the subtype through
+     * {@link JsonTypeInfo} on the {@code type} property, so {@code copyWithId(stringInput, ...) instanceof StringInput}.
+     */
+    private static Input<?> copyWithId(Input<?> input, String newId) {
+        Map<String, Object> map = JacksonMapper.toMap(input);
+        map.put("id", newId);
+        return JacksonMapper.toMap(map, Input.class);
+    }
+
+    /**
+     * @return all expanded leaf paths (e.g. {@code environment.region}, {@code credentials.api_key}, {@code api_key}),
+     * used by uniqueness validation to reject duplicate paths and prefix conflicts.
+     */
+    public static List<String> collectExpandedPaths(List<Input<?>> inputs) {
+        List<Input<?>> expanded = expandToLeaves(inputs);
+        return expanded == null ? List.of() : expanded.stream().map(Input::getId).toList();
+    }
 }

core/src/main/java/io/kestra/core/models/flows/Type.java

@@ -22,7 +22,8 @@ public enum Type {
     ARRAY(ArrayInput.class.getName()),
     MULTISELECT(MultiselectInput.class.getName()),
     YAML(YamlInput.class.getName()),
-    EMAIL(EmailInput.class.getName());
+    EMAIL(EmailInput.class.getName()),
+    FORM(FormInput.class.getName());
 
     private final String clsName;
 

core/src/main/java/io/kestra/core/models/flows/input/FormInput.java

@@ -0,0 +1,34 @@
+package io.kestra.core.models.flows.input;
+
+import java.util.List;
+
+import io.kestra.core.models.flows.Input;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.ConstraintViolationException;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder
+@Getter
+@NoArgsConstructor
+public class FormInput extends Input<Void> {
+    @Schema(
+        title = "The inputs grouped under this form.",
+        description = "A FORM groups related inputs under a shared display name. Each child input is resolved " +
+            "under the form's id as a nested path — e.g. a `region` input inside an `environment` form is " +
+            "referenced as `{{ inputs.environment.region }}`. To depend on a sibling, use the full dotted path " +
+            "in `dependsOn` (e.g. `dependsOn: [environment.data_center]`). Forms cannot be nested."
+    )
+    @NotNull
+    @Valid
+    private List<Input<?>> inputs;
+
+    @Override
+    public void validate(Void input) throws ConstraintViolationException {
+        // no-op: a FORM is a structural wrapper expanded into its children before resolution
+    }
+}

core/src/main/java/io/kestra/core/runners/FlowInputOutput.java

@@ -139,7 +139,9 @@ public Mono<Map<String, Object>> readExecutionInputs(final List<Input<?>> inputs
         return readData(inputs, execution, data, true).map(inputData -> this.readExecutionInputs(inputs, flow, execution, inputData));
     }
 
-    private Mono<Map<String, Object>> readData(List<Input<?>> inputs, Execution execution, Publisher<CompletedPart> data, boolean uploadFiles) {
+    private Mono<Map<String, Object>> readData(List<Input<?>> rawInputs, Execution execution, Publisher<CompletedPart> data, boolean uploadFiles) {
+        // Expand FORM inputs so FILE part matching works against dotted leaf ids.
+        final List<Input<?>> inputs = Input.expandToLeaves(rawInputs);
         return Flux.from(data)
             .publishOn(Schedulers.boundedElastic()).<Map.Entry<String, String>> handle((input, sink) ->
             {
@@ -281,8 +283,12 @@ public List<InputAndValue> resolveInputs(
             return Collections.emptyList();
         }
 
+        // Expand FORM inputs into dotted-id leaves so resolution runs on a flat list and the nested
+        // payload reassembles via flattenToNestedMap. Idempotent on already-expanded leaves.
+        final List<Input<?>> leafInputs = Input.expandToLeaves(inputs);
+
         final Map<String, ResolvableInput> resolvableInputMap = Collections.unmodifiableMap(
-            inputs.stream()
+            leafInputs.stream()
                 .map(input -> ResolvableInput.of(input, data.get(input.getId())))
                 .collect(Collectors.toMap(it -> it.get().input().getId(), Function.identity(), (o1, o2) -> o1, LinkedHashMap::new))
         );
@@ -333,23 +339,35 @@ private InputAndValue resolveInputValue(
                 return resolvable.get();
             }
 
-            // render input
-            input = RenderableInput.mayRenderInput(input, expression ->
-            {
-                try {
-                    return runContext.renderTyped(expression);
-                } catch (IllegalVariableEvaluationException e) {
-                    throw new RuntimeException(e.getMessage(), e);
-                }
-            });
+            // render input (e.g. a SELECT's dynamic `expression` values). A failure here means the field
+            // itself can't be rendered, so flag it as a render error so the UI can surface it eagerly.
+            try {
+                input = RenderableInput.mayRenderInput(input, expression ->
+                {
+                    try {
+                        return runContext.renderTyped(expression);
+                    } catch (IllegalVariableEvaluationException e) {
+                        throw new RuntimeException(e.getMessage(), e);
+                    }
+                });
+            } catch (Exception e) {
+                resolvable.resolveWithError(InputOutputValidationException.ofRenderError(e.getMessage(), input));
+                return resolvable.get();
+            }
             resolvable.setInput(input);
 
             Object value = resolvable.get().value();
 
-            // resolve default if needed
+            // resolve default if needed; a `defaults` that is a Pebble expression (e.g. subflow()/secret())
+            // can itself fail to render — that is also a broken field, so flag it as a render error.
             if (value == null && input.getDefaults() != null) {
                 RunContext runContextForDefault = decryptSecrets ? runContext : buildRunContextForExecutionAndInputs(flow, execution, dependencies, false);
-                value = resolveDefaultValue(input, runContextForDefault);
+                try {
+                    value = resolveDefaultValue(input, runContextForDefault);
+                } catch (Exception e) {
+                    resolvable.resolveWithError(InputOutputValidationException.ofRenderError(e.getMessage(), input));
+                    return resolvable.get();
+                }
                 resolvable.isDefault(true);
             }
 
@@ -396,6 +414,7 @@ public static Object resolveDefaultValue(Input<?> input, PropertyContext rendere
             case JSON, YAML -> resolveDefaultPropertyAs(input, renderer, Object.class);
             case ARRAY -> resolveDefaultPropertyAsList(input, renderer, Object.class);
             case MULTISELECT -> resolveDefaultPropertyAsList(input, renderer, String.class);
+            case FORM -> throw new IllegalStateException("FORM inputs must be expanded before resolution");
         };
     }
 
@@ -410,20 +429,21 @@ private static <T> Object resolveDefaultPropertyAsList(Input<?> input, PropertyC
     }
 
     private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies, final boolean decryptSecrets) {
-        Map<String, Object> flattenInputs = MapUtils.flattenToNestedMap(
-            dependencies.entrySet()
-                .stream()
-                .collect(HashMap::new, (m, v) -> m.put(v.getKey(), v.getValue().value()), HashMap::putAll)
-        );
+        Map<String, Object> flatInputs = dependencies.entrySet()
+            .stream()
+            .collect(HashMap::new, (m, v) -> m.put(v.getKey(), v.getValue().value()), HashMap::putAll);
         // Hack: Pre-inject all inputs that have a default value with 'null' to prevent
         // RunContextFactory from attempting to render them when absent, which could
         // otherwise cause an exception if a Pebble expression is involved.
-        List<Input<?>> inputs = Optional.ofNullable(flow).map(FlowInterface::getInputs).orElse(List.of());
+        // FORM inputs are expanded to dotted leaves first, and defaults are injected into the flat map
+        // before nesting so they end up under the form key (e.g. inputs.environment.region).
+        List<Input<?>> inputs = flow == null ? List.of() : flow.resolvableInputs();
         for (Input<?> input : inputs) {
-            if (input.getDefaults() != null && !flattenInputs.containsKey(input.getId())) {
-                flattenInputs.put(input.getId(), null);
+            if (input.getDefaults() != null && !flatInputs.containsKey(input.getId())) {
+                flatInputs.put(input.getId(), null);
             }
         }
+        Map<String, Object> flattenInputs = MapUtils.flattenToNestedMap(flatInputs);
         return runContextFactory.get().of(flow, execution, vars -> vars.withInputs(flattenInputs), decryptSecrets);
     }
 
@@ -558,6 +578,7 @@ private Object parseType(Execution execution, Type type, String id, Type element
                         yield asList;
                     }
                 }
+                case FORM -> throw new IllegalStateException("FORM inputs must be expanded before resolution");
             };
         } catch (IllegalArgumentException | ConstraintViolationException e) {
             throw e;

core/src/main/java/io/kestra/core/runners/RunContextFactory.java

@@ -254,7 +254,8 @@ private List<String> secretInputsFromFlow(FlowInterface flow) {
             return Collections.emptyList();
         }
 
-        return flow.getInputs().stream()
+        // FORM inputs are expanded to dotted leaves so a SECRET grouped under a form is masked by its dotted path.
+        return flow.resolvableInputs().stream()
             .filter(input -> input.getType() == Type.SECRET)
             .map(input -> input.getId()).toList();
     }