FORK_JOIN_DYNAMIC adds unexpected `input` nesting when using JSON_JQ_TRANSFORM output

[amphasis]

There seems to be a discrepancy in how Conductor OSS handles task inputs compared to Orkes Cloud, specifically in workflows that use FORK_JOIN_DYNAMIC together with JSON_JQ_TRANSFORM or INLINE tasks.

Expected Behavior (Orkes Cloud)

When providing forkTaskInputs in the following form:

[
  { "param1": "value", "param2": "value" },
  { "param1": "value", "param2": "value" }
]

each sub-workflow receives its input as-is:

{
  "param1": "value",
  "param2": "value"
}

Actual Behavior (Conductor OSS)

In Conductor OSS, when forkTaskInputs comes directly from the output of a JSON_JQ_TRANSFORM or INLINE task, the input to each sub-workflow is wrapped in an additional input field:

{
  "input": {
    "param1": "value",
    "param2": "value"
  }
}

Additional Findings

This behavior only happens when the output of JSON_JQ_TRANSFORM or INLINE is used directly by FORK_JOIN_DYNAMIC. If any asynchronous task (such as WAIT, HUMAN, SIMPLE, etc.) is placed in between, the behavior changes: the wrapping disappears, and the parameters are passed correctly.
This suggests that the issue occurs when Conductor OSS passes the raw output of a JSON_JQ_TRANSFORM (or INLINE) task directly into FORK_JOIN_DYNAMIC. In this case, each element gets wrapped with an additional input field. However, when an asynchronous task is placed in between, the workflow engine serializes and deserializes the intermediate state, which corrects the structure and removes the unwanted nesting.

Minimal workflow definition to reproduce:

{
  "name": "fork_join_dynamic_bug",
  "description": "A minimal reproduction WF for Conductor OSS FORK_JOIN_DYNAMIC bug",
  "version": 1,
  "tasks": [
    {
      "name": "json_transform",
      "taskReferenceName": "json_transform",
      "inputParameters": {
        "forkedProperty": [
          "value1",
          "value2"
        ],
        "queryExpression": ".forkedProperty[] as $property | {property: $property, commonProperty: .commonProperty}",
        "commonProperty": "common value"
      },
      "type": "JSON_JQ_TRANSFORM"
    },
    {
      "name": "fork_join_dynamic",
      "taskReferenceName": "fork_join_dynamic",
      "inputParameters": {
        "forkTaskWorkflow": "sub_workflow_definition_name",
        "forkTaskWorkflowVersion": "1",
        "forkTaskInputs": "${json_transform.output.resultList}"
      },
      "type": "FORK_JOIN_DYNAMIC",
      "dynamicForkTasksParam": "dynamicTasks",
      "dynamicForkTasksInputParamName": "dynamicTasksInput"
    },
    {
      "name": "join",
      "taskReferenceName": "join",
      "inputParameters": {},
      "type": "JOIN",
    }
  ],
  "inputParameters": [],
  "outputParameters": {},
  "schemaVersion": 2,
  "ownerEmail": "owner@email.com"
}

[sc-a-zhukov]

Also faced this problem. Below are the results of the research, but it would be nice if one of the maintainers validated it.

Where the extra "input" wrapping comes from

In FORK_JOIN_DYNAMIC, when building each forked task, OSS wraps any fork input that is not a Java Map into an input field
in the ForkJoinDynamicTaskMapper.java

if (forkTaskInput instanceof Map) {
    inputParameters.putAll((Map<? extends String, ?>) forkTaskInput);
} else {
    inputParameters.put("input", forkTaskInput);
}

The same logic exists for sub-workflows (from here:

if (forkTaskInput instanceof Map) {
	inputParameters.putAll((Map<? extends String, ?>) forkTaskInput);
	Map<? extends String, ?> forkTaskInputMap = (Map<? extends String, ?>) forkTaskInput;
	subWorkflowParams.setTaskToDomain(
			(Map<String, String>) forkTaskInputMap.get("taskToDomain"));
} else {
	inputParameters.put("input", forkTaskInput);
}

And SUB_WORKFLOW uses that map as the sub-workflow input (unless workflowInput is explicitly set) link:

var wfInput = (Map<String, Object>) input.get("workflowInput");
if (wfInput == null || wfInput.isEmpty()) {
	wfInput = input;
}

So if a fork element isn’t a Map, the sub-workflow sees {"input": ...}.

Probable cause why JSON_JQ_TRANSFORM and INLINE trigger it

JSON_JQ_TRANSFORM populates resultList with raw JsonNode values (not converted to Java Maps/Lists). Only the first result is converted via extractBody; resultList is not:
link JsonJqTransform.java

task.setStatus(TaskModel.Status.COMPLETED);
if (result == null) {
	task.addOutput(OUTPUT_RESULT, null);
	task.addOutput(OUTPUT_RESULT_LIST, null);
} else if (result.isEmpty()) {
	task.addOutput(OUTPUT_RESULT, null);
	task.addOutput(OUTPUT_RESULT_LIST, result);
} else {
	task.addOutput(OUTPUT_RESULT, extractBody(result.get(0)));
	task.addOutput(OUTPUT_RESULT_LIST, result);
}	

INLINE returns the evaluator result as-is (e.g., Nashorn ScriptObjectMirror), which is also not a Map:
link

Object evalResult = evaluator.evaluate(expression, taskInput);
task.addOutput("result", evalResult);
task.setStatus(TaskModel.Status.COMPLETED);

When these raw types are fed directly into FORK_JOIN_DYNAMIC as forkTaskInputs, the instanceof Map check fails, so OSS wraps each element under "input". If an async task runs in between, the output is serialized/deserialized and becomes plain Java Map/List, so the wrapping disappears.

Fix options (OSS)

• JSON_JQ_TRANSFORM: convert resultList elements to plain Java values before storing.
• ForkJoinDynamicTaskMapper: try coercing non-Map fork inputs into Map<String,Object> before defaulting to "input", e.g., attempt objectMapper.convertValue(forkTaskInput, new TypeReference<Map<String,Object>>() {}) and use it if successful.
• INLINE (optional hardening): convert result to plain Java types when it’s a script engine object.

Root cause

FORK_JOIN_DYNAMIC wraps any non-Map fork input; JSON_JQ_TRANSFORM.resultList and INLINE.result are non-Map (JsonNode/ScriptObjectMirror) until an async serialization step converts them.