fix(workflow): isolate PARALLEL_SPLIT branches and merge results explicitly

workflow/admin_recovery.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 
+	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 
 	"github.com/OpenNSW/core/shared/maputil"
@@ -23,11 +24,26 @@ type terminalAdminError struct {
 func (e *terminalAdminError) Error() string { return e.err.Error() }
 func (e *terminalAdminError) Unwrap() error { return e.err }
 
+// terminalAdminErrorType is the Go type name Temporal's default error converter stamps onto
+// the ApplicationError it wraps a plain (non-Temporal) error in when that error crosses a
+// child workflow boundary — e.g. a PARALLEL_SPLIT or BATCH_SPLIT branch's terminalAdminError,
+// returned from the child, re-materializes in the parent as *temporal.ApplicationError with
+// this Type(), not as a *terminalAdminError errors.As can find. Must match the unqualified
+// type name of terminalAdminError exactly.
+const terminalAdminErrorType = "terminalAdminError"
+
 // isTerminalAdminError reports whether err has already been through parkNodeForAdmin and
-// was deliberately given up on, meaning it should propagate without being parked again.
+// was deliberately given up on, meaning it should propagate without being parked again. This
+// checks both forms: a same-workflow error still holding its original Go type, and one that
+// crossed a child workflow boundary (a spawned PARALLEL_SPLIT or BATCH_SPLIT branch) and so
+// only carries the type name as a string — see terminalAdminErrorType.
 func isTerminalAdminError(err error) bool {
 	var terminal *terminalAdminError
-	return errors.As(err, &terminal)
+	if errors.As(err, &terminal) {
+		return true
+	}
+	var appErr *temporal.ApplicationError
+	return errors.As(err, &appErr) && appErr.Type() == terminalAdminErrorType
 }
 
 // AdminResolutionAction describes how an admin chooses to resolve a node that is
@@ -45,8 +61,7 @@ const (
 	AdminActionOverride AdminResolutionAction = "OVERRIDE"
 	// AdminActionSkip marks the node completed without setting any variables.
 	AdminActionSkip AdminResolutionAction = "SKIP"
-	// AdminActionAbort fails the node and the workflow with the original error —
-	// the same behavior the engine had before the escape hatch existed.
+	// AdminActionAbort fails the node and the workflow with the original error.
 	AdminActionAbort AdminResolutionAction = "ABORT"
 )
 

workflow/admin_recovery_test.go

@@ -380,6 +380,12 @@ func TestAdminSkipAndOverrideRejectedForParkedGatewayNode(t *testing.T) {
 // block a sibling branch running in parallel: the sibling completes while the first branch
 // is still NodeStatusAwaitingAdmin. Aborting the parked branch afterward still fails the
 // overall workflow, since parallel join semantics are unchanged.
+// TestAdminParkingIsolatesParallelBranches exercises PARALLEL_SPLIT's isolation: each
+// matching branch runs as its own child workflow (see parallel_gateway.go), so a node parked
+// for admin inside one branch (task_a, here) shows up in that child's own status and signal
+// channel, not the parent's (similar to BATCH_SPLIT partitions).
+// The child's deterministic ID is FormatBatchChildWorkflowID(parentID, splitNodeID, edgeID);
+// parallelWorkflowJSON's split->task_a edge is "e2".
 func TestAdminParkingIsolatesParallelBranches(t *testing.T) {
 	testSuite := &testsuite.WorkflowTestSuite{}
 	env := testSuite.NewTestWorkflowEnvironment()
@@ -396,21 +402,22 @@ func TestAdminParkingIsolatesParallelBranches(t *testing.T) {
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_B", mock.Anything).
 		Return(map[string]any{}, nil).Once()
 
+	branchWorkflowID := FormatBatchChildWorkflowID("default-test-workflow-id", "split", "e2")
+
 	env.RegisterDelayedCallback(func() {
-		val, err := env.QueryWorkflow("GetStatus")
+		val, err := env.QueryWorkflowByID(branchWorkflowID, "GetStatus")
 		require.NoError(t, err)
 		var instance WorkflowInstance
 		require.NoError(t, val.Get(&instance))
 
 		require.Equal(t, NodeStatusAwaitingAdmin, instance.NodeInfo["task_a"].Status)
-		require.Equal(t, NodeStatusCompleted, instance.NodeInfo["task_b"].Status)
 	}, time.Second)
 
 	env.RegisterDelayedCallback(func() {
-		env.SignalWorkflow(AdminResolutionSignalName, AdminResolutionSignal{
+		require.NoError(t, env.SignalWorkflowByID(branchWorkflowID, AdminResolutionSignalName, AdminResolutionSignal{
 			NodeID: "task_a",
 			Action: AdminActionAbort,
-		})
+		}))
 	}, 2*time.Second)
 
 	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{})

workflow/batch_gateway_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/suite"
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/testsuite"
 	"go.temporal.io/sdk/workflow"
 )
@@ -1332,3 +1333,79 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildReturnsDuplicateItemIDAcross
 	s.Error(err)
 	s.Contains(err.Error(), "duplicate item ID \"item-a\" returned across child partitions")
 }
+
+// --- Test 16: Child workflow task aborted by admin propagates to parent without re-parking ---
+
+func (s *BatchGatewayTestSuite) TestBatchSplit_ChildTaskAdminAbort_PropagatesWithoutReparkingParent() {
+	env := s.NewTestWorkflowEnvironment()
+
+	acts := &Activities{}
+	env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"})
+	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
+
+	def := WorkflowDefinition{
+		ID:   "batch_child_abort_test",
+		Name: "Batch Child Abort Test",
+		Nodes: []Node{
+			{ID: "start", Type: NodeTypeStart},
+			{ID: "gw_split", Type: NodeTypeGateway, GatewayType: GatewayTypeBatchSplit,
+				BatchGateway: &BatchGatewayConfig{}}, // defaults: _items, id
+			{ID: "process", Type: NodeTypeTask, TaskTemplateID: "PROCESS"},
+			{ID: "gw_join", Type: NodeTypeGateway, GatewayType: GatewayTypeBatchJoin,
+				BatchJoin: &BatchJoinConfig{GatewayNodeID: "gw_split"}},
+			{ID: "post_task", Type: NodeTypeTask, TaskTemplateID: "POST_TASK"},
+			{ID: "end", Type: NodeTypeEnd},
+		},
+		Edges: []Edge{
+			{ID: "e1", SourceID: "start", TargetID: "gw_split"},
+			{ID: "e2", SourceID: "gw_split", TargetID: "process", Condition: `item.needsWork == true`},
+			{ID: "e3", SourceID: "process", TargetID: "gw_join"},
+			{ID: "e4", SourceID: "gw_join", TargetID: "post_task"},
+			{ID: "e5", SourceID: "post_task", TargetID: "end"},
+		},
+	}
+
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS", mock.Anything).
+		Return(nil, temporal.NewNonRetryableApplicationError("inspection boom", "TaskFailure", nil)).Once()
+
+	parentWorkflowID := "batch-child-abort-1"
+	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
+	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: parentWorkflowID})
+
+	childWorkflowID := FormatBatchChildWorkflowID(parentWorkflowID, "gw_split", "e2")
+
+	// 1. Verify that the child's node is parked awaiting admin intervention
+	env.RegisterDelayedCallback(func() {
+		val, err := env.QueryWorkflowByID(childWorkflowID, "GetStatus")
+		s.Require().NoError(err)
+		var instance WorkflowInstance
+		s.Require().NoError(val.Get(&instance))
+		s.Require().NotNil(instance.NodeInfo["process"], "node 'process' must exist in child NodeInfo")
+		s.Equal(NodeStatusAwaitingAdmin, instance.NodeInfo["process"].Status)
+	}, time.Second)
+
+	// 2. Resolve the child node with AdminActionAbort
+	env.RegisterDelayedCallback(func() {
+		s.NoError(env.SignalWorkflowByID(childWorkflowID, AdminResolutionSignalName, AdminResolutionSignal{
+			NodeID: "process",
+			Action: AdminActionAbort,
+		}))
+	}, 2*time.Second)
+
+	initialVars := map[string]any{
+		"_items": []any{
+			map[string]any{"id": "item1", "needsWork": true},
+		},
+	}
+
+	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, initialVars)
+
+	// Workflow completes with failure immediately without re-parking the parent's gw_split
+	s.True(env.IsWorkflowCompleted())
+	err := env.GetWorkflowError()
+	s.Error(err)
+	s.Contains(err.Error(), "inspection boom")
+
+	// Verify post_task was never invoked
+	env.AssertNotCalled(s.T(), "ExecuteTaskActivity", mock.Anything, "POST_TASK", mock.Anything)
+}

workflow/batch_validate.go

@@ -68,28 +68,36 @@ func ValidateBatchGateways(def WorkflowDefinition) error {
 	// 4. Sub-graph topological containment: enforce that every path from BATCH_SPLIT
 	// reaches its paired BATCH_JOIN, and no edges escape or illegally enter the sub-graph region.
 	for splitID, joinID := range pairedJoins {
-		if err := validateBatchRegion(splitID, joinID, pairedJoins, nodesByID, forwardEdges, reverseEdges); err != nil {
+		if err := validateGatewayRegion(splitID, joinID, GatewayTypeBatchSplit, "BATCH_SPLIT", "BATCH_JOIN", "batch region", pairedJoins, nodesByID, forwardEdges, reverseEdges); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-// validateBatchRegion checks that the sub-graph between splitID and joinID is strictly
-// closed: all paths must terminate at joinID, with no dead ends, escaped edges, or illegal entries.
-func validateBatchRegion(
+// validateGatewayRegion checks that the sub-graph between splitID and joinID is strictly
+// closed: all paths must terminate at joinID, with no dead ends, escaped edges, or illegal
+// entries. Shared by ValidateBatchGateways and ValidateParallelGateways — nestedType and
+// pairedJoins scope the "nested split's join must stay inside this region" check to gateways
+// of the same kind as splitID itself (a nested BATCH_SPLIT inside a BATCH_SPLIT region, or a
+// nested PARALLEL_SPLIT inside a PARALLEL_SPLIT region); it does not check cross-kind nesting.
+// splitLabel/joinLabel/regionLabel format the error messages (e.g. "BATCH_SPLIT"/"BATCH_JOIN"/
+// "batch region" vs "PARALLEL_SPLIT"/"PARALLEL_JOIN"/"parallel region").
+func validateGatewayRegion(
 	splitID, joinID string,
+	nestedType GatewayType,
+	splitLabel, joinLabel, regionLabel string,
 	pairedJoins map[string]string,
 	nodesByID map[string]*Node,
 	forwardEdges map[string][]Edge,
 	reverseEdges map[string][]string,
 ) error {
 	if len(forwardEdges[splitID]) == 0 {
-		return fmt.Errorf("BATCH_SPLIT node %q has no outgoing edges", splitID)
+		return fmt.Errorf("%s node %q has no outgoing edges", splitLabel, splitID)
 	}
 
-	// 1. Forward reachability: find all nodes in the batch region, stopping at joinID.
+	// 1. Forward reachability: find all nodes in the region, stopping at joinID.
 	regionNodes := make(map[string]bool)
 	queue := []string{splitID}
 	regionNodes[splitID] = true
@@ -132,40 +140,40 @@ func validateBatchRegion(
 	for nodeID := range regionNodes {
 		node, exists := nodesByID[nodeID]
 		if !exists {
-			return fmt.Errorf("BATCH_SPLIT node %q region references non-existent node %q", splitID, nodeID)
+			return fmt.Errorf("%s node %q region references non-existent node %q", splitLabel, splitID, nodeID)
 		}
 		if node.Type == NodeTypeEnd {
-			return fmt.Errorf("BATCH_SPLIT node %q has a path reaching END node %q without passing through paired BATCH_JOIN %q", splitID, nodeID, joinID)
+			return fmt.Errorf("%s node %q has a path reaching END node %q without passing through paired %s %q", splitLabel, splitID, nodeID, joinLabel, joinID)
 		}
 		if len(forwardEdges[nodeID]) == 0 {
-			return fmt.Errorf("BATCH_SPLIT node %q region contains dead-end node %q with no outgoing edges", splitID, nodeID)
+			return fmt.Errorf("%s node %q region contains dead-end node %q with no outgoing edges", splitLabel, splitID, nodeID)
 		}
 		if !canReachJoin[nodeID] {
-			return fmt.Errorf("BATCH_SPLIT node %q region contains node %q which cannot reach paired BATCH_JOIN %q", splitID, nodeID, joinID)
+			return fmt.Errorf("%s node %q region contains node %q which cannot reach paired %s %q", splitLabel, splitID, nodeID, joinLabel, joinID)
 		}
 
-		// Nested BATCH_SPLIT must have its paired BATCH_JOIN contained within this batch region.
-		if node.Type == NodeTypeGateway && node.GatewayType == GatewayTypeBatchSplit && nodeID != splitID {
+		// A nested split of the SAME kind must have its paired join contained within this region.
+		if node.Type == NodeTypeGateway && node.GatewayType == nestedType && nodeID != splitID {
 			nestedJoinID := pairedJoins[nodeID]
 			if !regionNodes[nestedJoinID] {
-				return fmt.Errorf("BATCH_SPLIT node %q contains nested BATCH_SPLIT %q whose paired BATCH_JOIN %q is outside the batch region", splitID, nodeID, nestedJoinID)
+				return fmt.Errorf("%s node %q contains nested %s %q whose paired %s %q is outside the %s", splitLabel, splitID, splitLabel, nodeID, joinLabel, nestedJoinID, regionLabel)
 			}
 		}
 
-		// Encapsulation: no edges from outside the batch region may enter intermediate region nodes.
+		// Encapsulation: no edges from outside the region may enter intermediate region nodes.
 		if nodeID != splitID {
 			for _, prev := range reverseEdges[nodeID] {
 				if !regionNodes[prev] {
-					return fmt.Errorf("BATCH_SPLIT node %q region contains node %q with incoming edge from outside the batch region (from %q)", splitID, nodeID, prev)
+					return fmt.Errorf("%s node %q region contains node %q with incoming edge from outside the %s (from %q)", splitLabel, splitID, nodeID, regionLabel, prev)
 				}
 			}
 		}
 	}
 
-	// 4. Encapsulation: all incoming edges to the paired BATCH_JOIN must originate within this batch region.
+	// 4. Encapsulation: all incoming edges to the paired join must originate within this region.
 	for _, prev := range reverseEdges[joinID] {
 		if !regionNodes[prev] {
-			return fmt.Errorf("BATCH_JOIN node %q has incoming edge from node %q outside its paired BATCH_SPLIT %q region", joinID, prev, splitID)
+			return fmt.Errorf("%s node %q has incoming edge from node %q outside its paired %s %q region", joinLabel, joinID, prev, splitLabel, splitID)
 		}
 	}
 

workflow/dsl.go

@@ -161,6 +161,7 @@ type Node struct {
 	Signaling    *SignalingConfig    `json:"signaling,omitempty"`
 	BatchGateway *BatchGatewayConfig `json:"batch_gateway,omitempty"`
 	BatchJoin    *BatchJoinConfig    `json:"batch_join,omitempty"`
+	ParallelJoin *ParallelJoinConfig `json:"parallel_join,omitempty"`
 }
 
 // Edge represents a directed connection between two nodes.
@@ -219,6 +220,59 @@ type BatchJoinConfig struct {
 	IDField string `json:"id_field,omitempty"`
 }
 
+// ParallelJoinConfig configures how a PARALLEL_JOIN gateway isolates its branches and
+// merges their final workflow variable states back into the parent scope.
+//
+// WARNING: CONFLICT RESOLUTION AND LAST-WRITE-WINS (LWW)
+// When concurrent parallel branches write conflicting data, the engine cannot infer intent
+// and applies a deterministic "last-write-wins" policy. Precedence is determined strictly by
+// branch edge order (sorted ascending by source edge ID; later branches in that order overwrite
+// earlier ones), NOT wall-clock completion time.
+//
+// Specifically, LAST-WRITE-WINS applies in the following three places:
+//  1. Conflicting item fields in MergeByID: If two branches modify the SAME field on the SAME
+//     item (e.g. branch A sets item["status"]="PASS" and branch B sets item["status"]="FAIL"),
+//     the later branch's field value overwrites the earlier one.
+//  2. Conflicting map keys: If two branches set the SAME leaf key in a map[string]any, the
+//     later branch's value overwrites the earlier one.
+//  3. Scalars and unlisted arrays: Any scalar variable (string, number, boolean) or slice not
+//     listed in MergeByID that is written by multiple branches will be completely overwritten
+//     by the later branch.
+//
+// Recommendation for workflow authors: Ensure concurrent parallel branches write to distinct
+// variable paths or distinct field names on shared items if overwriting cannot be tolerated.
+//
+// Execution and Reconciliation Model:
+// Each matching outgoing edge of the paired PARALLEL_SPLIT runs as its own child workflow
+// with a deep-copied, isolated set of WorkflowVariables — no branch can see another
+// branch's writes while any of them are still running. When all branches complete,
+// PARALLEL_JOIN reconciles their (possibly divergent) final states back into one:
+//
+//   - map[string]any values are deep-merged key by key (recursive map merge):
+//     each branch's mutated sub-fields combine, so two branches changing different
+//     fields of the same object both survive.
+//   - Variables listed in MergeByID (each a []map[string]any) are merged element-by-element
+//     by that id_field across ALL branches — not "one branch's array replaces another's":
+//     for a given item ID, the fields each branch's copy of that item carries are unioned
+//     into one item, so lab writing sample_test_result and visual writing visual_result to
+//     the same underlying item both survive regardless of which branch finishes last. A
+//     field two branches BOTH carry a (possibly different) value for is a genuine conflict
+//     this cannot resolve: branches are applied in a fixed, deterministic order (sorted by
+//     source edge ID) and the last one wins for that field — give each branch its own field
+//     name if that ambiguity isn't acceptable.
+//   - Any other variable (scalar, or array not listed in MergeByID) falls back to the same
+//     fixed deterministic branch order, last one wins.
+type ParallelJoinConfig struct {
+	// GatewayNodeID is the node ID of the paired PARALLEL_SPLIT gateway.
+	GatewayNodeID string `json:"gateway_node_id"`
+
+	// MergeByID maps a top-level workflow variable name (holding a []map[string]any shared by
+	// multiple branches) to the field name within each item used as its unique ID.
+	// Nested dot-paths are not currently supported; see ValidateParallelGateways.
+	// TODO: Support nested dot-paths (e.g. "order.items") in MergeByID and mergeVariablesInto.
+	MergeByID map[string]string `json:"merge_by_id,omitempty"`
+}
+
 // VarScopePath is the workflow variable key holding the hierarchical scope path string
 // for batch gateway nesting (e.g. "root/gw_type/lab/gw_result/fail").
 const VarScopePath = "_scope_path"