feat(DurableExecution): incremental, heterogeneous Parallel API (#2519)

.autover/changes/12a4a1f7-d59f-4544-aa1f-6db30289485e.json

@@ -2,7 +2,7 @@
   "Projects": [
     {
       "Name": "Amazon.Lambda.DurableExecution",
-      "Type": "Minor",
+      "Type": "Major",
       "ChangelogMessages": [
         "Behavior change: steps and child contexts now round-trip their result through the configured serializer on a fresh (non-replay) success, deserializing the just-written checkpoint before returning \u2014 matching replay semantics. A custom per-operation serializer\u0027s transform is now reflected in the operation result on the first execution, not only on replay. The returned value is a fresh deserialized instance rather than the exact object the step/child body produced. Overflow (replay-children) results are unaffected."
       ]

.autover/changes/add-incremental-heterogeneous-parallel.json

@@ -0,0 +1,12 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Minor",
+      "ChangelogMessages": [
+        "Added an incremental, branch-oriented parallel API (IDurableContext.CreateParallel, IDurableParallel, IParallelBranch<T>) supporting heterogeneous per-branch result types and incremental branch registration, alongside the existing homogeneous ParallelAsync<T> overloads. Each branch declares its own result type and returns an awaitable typed handle; branches start on registration (respecting MaxConcurrency) and the operation is sealed with CompleteAsync. Honors the existing MaxConcurrency, CompletionConfig, NestingType, cancellation, deterministic replay, and ILambdaSerializer behavior.",
+        "CreateParallel supports per-operation and per-branch result serialization: ParallelConfig.ItemSerializer sets the operation-level serializer for all branch results, and IDurableParallel.Branch accepts an optional per-branch ILambdaSerializer override, falling back to ItemSerializer and then the globally-registered serializer."
+      ]
+    }
+  ]
+}

Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs

@@ -192,6 +192,29 @@ private Task<ICallback<T>> RunCallback<T>(
         return op.ExecuteAsync(cancellationToken);
     }
 
+    public IDurableParallel CreateParallel(
+        string? name = null,
+        ParallelConfig? config = null)
+    {
+        var effectiveConfig = config ?? new ParallelConfig();
+        // Operation-level default for per-branch result serialization: the config's
+        // ItemSerializer if set, else the globally-registered serializer. Individual
+        // branches may still override this via Branch's serializer parameter.
+        //
+        // Resolved LAZILY: a workflow that overrides the serializer on every Branch
+        // call (the AOT/per-branch scenario) must not be forced to register a global
+        // serializer. GetRequired is deferred to the factory below and invoked only
+        // when a branch actually falls back to this operation-level default.
+        var lambdaContext = LambdaContext;
+        Func<Amazon.Lambda.Core.ILambdaSerializer> defaultSerializerFactory =
+            () => effectiveConfig.ItemSerializer ?? LambdaSerializerHelper.GetRequired(lambdaContext);
+
+        var operationId = _idGenerator.NextId();
+        return new Internal.IncrementalParallelOperation(
+            operationId, name, _idGenerator.ParentId, effectiveConfig, defaultSerializerFactory, MakeChildFactory(),
+            _state, _terminationManager, _workflowCancellation, _durableExecutionArn, _batcher);
+    }
+
     public Task<IBatchResult<T>> ParallelAsync<T>(
         IReadOnlyList<Func<IDurableContext, CancellationToken, Task<T>>> branches,
         string? name = null,

Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs

@@ -367,6 +367,47 @@ Task<TState> WaitForConditionAsync<TState>(
         string? name = null,
         CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Create an incremental, branch-oriented parallel operation. Unlike the
+    /// <see cref="ParallelAsync{T}(IReadOnlyList{Func{IDurableContext, CancellationToken, Task{T}}}, string?, ParallelConfig?, CancellationToken)"/>
+    /// overloads — which take a complete branch list up front and share one result
+    /// type — the returned <see cref="IDurableParallel"/> lets you register branches
+    /// one at a time via
+    /// <see cref="IDurableParallel.Branch{T}"/>,
+    /// each with its own result type (heterogeneous), starting each branch as it is
+    /// registered. Call <see cref="IDurableParallel.CompleteAsync(CancellationToken)"/>
+    /// to seal registration and obtain the aggregate <see cref="IBatchResult"/>.
+    /// </summary>
+    /// <remarks>
+    /// Use <c>await using</c> so the operation is sealed and its terminal checkpoint
+    /// written even if <see cref="IDurableParallel.CompleteAsync(CancellationToken)"/>
+    /// is not called. Branch identity is positional and deterministic across
+    /// replays, so register the same branches in the same order every invocation —
+    /// derive any dynamic branch set from a checkpointed step. Per-branch results
+    /// are serialized via the <see cref="ILambdaSerializer"/> registered on
+    /// <see cref="ILambdaContext.Serializer"/>. Honors the same
+    /// <see cref="ParallelConfig.MaxConcurrency"/>,
+    /// <see cref="ParallelConfig.CompletionConfig"/>, and
+    /// <see cref="ParallelConfig.NestingType"/> as the homogeneous API.
+    /// </remarks>
+    /// <param name="name">
+    /// Optional human-readable name for the parallel operation, used only for
+    /// observability — it surfaces on the wire <c>OperationUpdate.Name</c> field and
+    /// in execution traces. The deterministic operation ID is positional (derived
+    /// from the call order, not from this name), so a name change across deployments
+    /// does not break replay. Defaults to <c>null</c>.
+    /// </param>
+    /// <param name="config">
+    /// Optional parallel configuration. Defaults are used when null.
+    /// </param>
+    /// <returns>
+    /// An <see cref="IDurableParallel"/> for registering branches and awaiting the
+    /// aggregate result.
+    /// </returns>
+    IDurableParallel CreateParallel(
+        string? name = null,
+        ParallelConfig? config = null);
+
     /// <summary>
     /// Execute multiple branches concurrently. Each branch runs inside its own
     /// child context; per-branch results are aggregated into an

Libraries/src/Amazon.Lambda.DurableExecution/IDurableParallel.cs

@@ -0,0 +1,130 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+namespace Amazon.Lambda.DurableExecution;
+
+/// <summary>
+/// An incremental, branch-oriented parallel operation created by
+/// <see cref="IDurableContext.CreateParallel(string?, ParallelConfig?)"/>. Branches
+/// are registered one at a time via
+/// <see cref="Branch{T}"/>,
+/// each with its own result type (heterogeneous), and each begins executing
+/// immediately (subject to <see cref="ParallelConfig.MaxConcurrency"/>). Call
+/// <see cref="CompleteAsync(System.Threading.CancellationToken)"/> to seal
+/// registration, await the branches according to the
+/// <see cref="ParallelConfig.CompletionConfig"/>, and obtain the aggregate result.
+/// </summary>
+/// <remarks>
+/// This is an additive alternative to the homogeneous
+/// <see cref="IDurableContext.ParallelAsync{T}(System.Collections.Generic.IReadOnlyList{System.Func{IDurableContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>
+/// overloads, which accept a complete branch list up front and share one result
+/// type. Use <c>CreateParallel</c> when branches return unrelated types, or when
+/// branches are discovered incrementally (for example, tool calls derived from a
+/// checkpointed plan) and earlier branches should start before later ones are known.
+/// <para>
+/// <b>Deterministic replay.</b> Branch identity is positional: the n-th
+/// <see cref="Branch{T}"/>
+/// call reuses the n-th deterministic operation ID. Workflow code must therefore
+/// register the same branches in the same order across invocations — produce any
+/// dynamic branch list inside a checkpointed <see cref="IDurableContext.StepAsync{T}(System.Func{IStepContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}, string?, StepConfig?, System.Threading.CancellationToken)"/>
+/// so replay sees the same set.
+/// </para>
+/// <para>
+/// <b>Disposal.</b> <see cref="System.IAsyncDisposable.DisposeAsync"/> seals and
+/// completes the operation if
+/// <see cref="CompleteAsync(System.Threading.CancellationToken)"/> was not called,
+/// so an <c>await using</c> block always writes the parallel's terminal checkpoint.
+/// Calling <see cref="CompleteAsync(System.Threading.CancellationToken)"/>
+/// explicitly is recommended so you can capture the aggregate result.
+/// </para>
+/// </remarks>
+public interface IDurableParallel : IAsyncDisposable
+{
+    /// <summary>
+    /// Registers a branch and immediately begins executing it (respecting
+    /// <see cref="ParallelConfig.MaxConcurrency"/>). Returns a typed
+    /// <see cref="IParallelBranch{T}"/> handle for retrieving the branch's result.
+    /// </summary>
+    /// <remarks>
+    /// The branch runs inside its own child context with a deterministic
+    /// operation-ID space; its result is serialized to a checkpoint via the
+    /// <see cref="Amazon.Lambda.Core.ILambdaSerializer"/> registered on
+    /// <see cref="Amazon.Lambda.Core.ILambdaContext.Serializer"/>. Per-branch
+    /// failures are captured on the handle and aggregated into the
+    /// <see cref="CompleteAsync(System.Threading.CancellationToken)"/> result — a
+    /// branch failure never throws out of this method.
+    /// </remarks>
+    /// <typeparam name="T">The branch's result type.</typeparam>
+    /// <param name="name">
+    /// Human-readable branch name. Required; surfaces on
+    /// <c>OperationUpdate.Name</c> and must remain stable at a given branch index
+    /// across deployments (a drift is a non-deterministic-execution error).
+    /// </param>
+    /// <param name="func">
+    /// The branch body. Receives its own <see cref="IDurableContext"/> and a
+    /// <see cref="System.Threading.CancellationToken"/> linking the SDK's
+    /// workflow-shutdown signal with the operation's completion-policy
+    /// short-circuit, and returns the branch's result.
+    /// </param>
+    /// <param name="serializer">
+    /// Optional serializer for <em>this branch's</em> result payload. When
+    /// <c>null</c> (default), the branch uses the operation-level serializer —
+    /// <see cref="ParallelConfig.ItemSerializer"/> if set on
+    /// <see cref="IDurableContext.CreateParallel(string?, ParallelConfig?)"/>, otherwise
+    /// the globally-registered <see cref="Amazon.Lambda.Core.ILambdaSerializer"/> on
+    /// <see cref="Amazon.Lambda.Core.ILambdaContext.Serializer"/>. Because each branch
+    /// declares its own result type, a per-branch serializer lets one branch use a
+    /// bespoke serializer (for example a source-generated context for AOT) without
+    /// affecting sibling branches. It is part of the deterministic definition: the same
+    /// branch must be able to deserialize a result it previously serialized, so pass the
+    /// same serializer at a given branch index across replays.
+    /// </param>
+    /// <returns>A typed handle for awaiting the branch's result.</returns>
+    /// <exception cref="System.ArgumentNullException">
+    /// <paramref name="name"/> or <paramref name="func"/> is <c>null</c>.
+    /// </exception>
+    /// <exception cref="System.InvalidOperationException">
+    /// The operation has already been sealed by
+    /// <see cref="CompleteAsync(System.Threading.CancellationToken)"/>.
+    /// </exception>
+    /// <exception cref="System.ObjectDisposedException">
+    /// The operation has already been disposed.
+    /// </exception>
+    /// <exception cref="NonDeterministicExecutionException">
+    /// On replay, the <paramref name="name"/> at this branch index differs from the
+    /// name recorded in the checkpoint for a previous invocation (branch name drift).
+    /// </exception>
+    IParallelBranch<T> Branch<T>(
+        string name,
+        Func<IDurableContext, CancellationToken, Task<T>> func,
+        Amazon.Lambda.Core.ILambdaSerializer? serializer = null);
+
+    /// <summary>
+    /// Seals registration (no further branches may be added), awaits the
+    /// registered branches according to the
+    /// <see cref="ParallelConfig.CompletionConfig"/>, checkpoints the aggregate
+    /// outcome, and returns it. Idempotent — repeated calls return the same result.
+    /// </summary>
+    /// <remarks>
+    /// Like the homogeneous parallel API, this never throws on per-branch failure:
+    /// inspect <see cref="IBatchResult.HasFailure"/> /
+    /// <see cref="IBatchResult.CompletionReason"/>, or await individual branch
+    /// handles, to observe failures. It does propagate workflow-level errors (for
+    /// example <see cref="NonDeterministicExecutionException"/>) and cancellation.
+    /// <para>
+    /// The <paramref name="cancellationToken"/> does not interrupt in-flight branch
+    /// settlement. Because branches begin executing when they are registered (before
+    /// <c>CompleteAsync</c> is called), this token is neither retroactively linked
+    /// into already-running branch bodies nor into the wait for them to settle —
+    /// those observe the SDK's workflow-shutdown signal (and the completion-policy
+    /// short-circuit) instead. This call blocks until every dispatched branch reaches
+    /// its terminal checkpoint; the token is observed only after that, so cancellation
+    /// is surfaced just before the aggregate result would be returned rather than
+    /// cutting the wait short. Dispatched branches always run to a terminal checkpoint
+    /// so replay stays deterministic, matching <see cref="IDurableContext.ParallelAsync{T}(System.Collections.Generic.IReadOnlyList{System.Func{IDurableContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>.
+    /// </para>
+    /// </remarks>
+    /// <param name="cancellationToken">A token to observe for cancellation.</param>
+    /// <returns>The aggregate <see cref="IBatchResult"/> summarizing branch outcomes.</returns>
+    Task<IBatchResult> CompleteAsync(CancellationToken cancellationToken = default);
+}

Libraries/src/Amazon.Lambda.DurableExecution/IParallelBranch.cs

@@ -0,0 +1,65 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+using System.Runtime.CompilerServices;
+
+namespace Amazon.Lambda.DurableExecution;
+
+/// <summary>
+/// A typed handle to a single branch registered on an <see cref="IDurableParallel"/>
+/// via <see cref="IDurableParallel.Branch{T}"/>.
+/// Unlike the homogeneous <see cref="IDurableContext.ParallelAsync{T}(System.Collections.Generic.IReadOnlyList{System.Func{IDurableContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>
+/// API — where every branch shares one result type <c>T</c> — each branch on an
+/// <see cref="IDurableParallel"/> declares its own result type, so a single
+/// parallel operation can mix, for example, an <c>InventoryReservation</c> branch
+/// with a <c>PaymentAuthorization</c> branch.
+/// </summary>
+/// <remarks>
+/// The handle is <c>await</c>-able: <c>await branch</c> yields the branch's typed
+/// result once it succeeds, or rethrows the branch's failure (a
+/// <see cref="ChildContextException"/>) if it failed. Awaiting a branch that was
+/// skipped by the operation's <see cref="CompletionConfig"/> short-circuit (its
+/// <see cref="Status"/> is <see cref="BatchItemStatus.Started"/>) throws a
+/// <see cref="DurableExecutionException"/> — inspect <see cref="Status"/> before
+/// awaiting when a completion policy may skip branches.
+/// <para>
+/// Typically you await the handle <em>after</em>
+/// <see cref="IDurableParallel.CompleteAsync(System.Threading.CancellationToken)"/>
+/// has sealed and resolved the operation, mirroring the Java SDK's
+/// <c>future.get()</c> after the <c>try</c>-with-resources block. A branch may
+/// still be awaited earlier; the await simply completes when the branch does.
+/// </para>
+/// </remarks>
+/// <typeparam name="T">The branch's result type.</typeparam>
+public interface IParallelBranch<T>
+{
+    /// <summary>
+    /// The branch name supplied at registration. Surfaces on the wire
+    /// <c>OperationUpdate.Name</c> field and in execution traces.
+    /// </summary>
+    string Name { get; }
+
+    /// <summary>
+    /// Zero-based registration order of this branch within its parallel
+    /// operation. Stable across replays. The branch's deterministic operation ID
+    /// is derived from the <em>one-based</em> position (<c>hash("{parentId}-{Index+1}")</c>),
+    /// so the first branch (<c>Index</c> 0) uses suffix 1.
+    /// </summary>
+    int Index { get; }
+
+    /// <summary>
+    /// The branch's outcome. <see cref="BatchItemStatus.Started"/> until the
+    /// branch settles (and permanently for a branch skipped by a completion-policy
+    /// short-circuit), then <see cref="BatchItemStatus.Succeeded"/> or
+    /// <see cref="BatchItemStatus.Failed"/>.
+    /// </summary>
+    BatchItemStatus Status { get; }
+
+    /// <summary>
+    /// Enables <c>await branch</c>. Yields the branch's typed result on success,
+    /// rethrows its <see cref="ChildContextException"/> on failure, or throws a
+    /// <see cref="DurableExecutionException"/> if the branch was skipped.
+    /// </summary>
+    /// <returns>An awaiter over the branch's result.</returns>
+    TaskAwaiter<T> GetAwaiter();
+}