Feature/durable result serializer

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

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "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/3691fc04-ad66-49c5-a383-f9592a6da0ce.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Major",
+      "ChangelogMessages": [
+        "Behavior change: when a step\u0027s fresh-success serializer round-trip fails (e.g. an asymmetric/broken StepConfig.Serializer that cannot deserialize its own output), the step now fails TERMINALLY (FAIL checkpoint \u002B StepException) instead of being routed through the configured RetryStrategy. Previously a non-null RetryStrategy would re-invoke the already-succeeded, side-effecting step body on each attempt (duplicating side effects) until attempts exhausted. This mirrors ChildContextOperation\u0027s terminal handling of a round-trip failure."
+      ]
+    }
+  ]
+}

.autover/changes/3d3d307e-cf33-421d-b9c4-c88670b5c5ef.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Minor",
+      "ChangelogMessages": [
+        "Add FileSystemSerializer for offloading large durable operation results to a filesystem (e.g. EFS/S3 Files), keeping only a small pointer in the checkpoint. It wraps an inner ILambdaSerializer, so callers control the on-the-wire format (JSON, compressed, etc.). Introduces the optional IDurableResultSerializer interface and DurableSerializationContext (EntityId \u002B DurableExecutionArn); when an operation\u0027s configured serializer implements it, the durable runtime supplies operation/execution identity so external storage can be keyed safely. Applied to Step, ChildContext, WaitForCondition, and Map/Parallel per-item results. Plain ILambdaSerializer serializers are unaffected."
+      ]
+    }
+  ]
+}

.autover/changes/45498f15-84f7-47cc-a871-47cdb059a0c4.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Minor",
+      "ChangelogMessages": [
+        "Add optional per-item serializer override (ItemSerializer) on MapConfig and ParallelConfig; when null the globally-registered ILambdaSerializer is used for item/branch results."
+      ]
+    }
+  ]
+}

.autover/changes/7ed69083-0a98-424b-8d30-3d70c916c2a1.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Minor",
+      "ChangelogMessages": [
+        "Add optional per-operation serializer override (Serializer) on StepConfig, CallbackConfig, InvokeConfig, WaitForConditionConfig, and ChildContextConfig; when null the globally-registered ILambdaSerializer is used."
+      ]
+    }
+  ]
+}

.autover/changes/c1cb8e3a-cca9-415e-b9ca-08289658df08.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Patch",
+      "ChangelogMessages": [
+        "Behavior change: when a step post-round-trip SUCCEED-checkpoint enqueue fails (e.g. a broken CheckpointBatcher), the failure is now routed through the terminal path (FAIL checkpoint plus StepException) rather than propagating raw, consistent with the fresh-success round-trip failure handling. The side-effecting body is never re-invoked, and if emitting the terminal FAIL itself throws, the error propagates without recursion."
+      ]
+    }
+  ]
+}

.autover/changes/ca5a9844-031a-4442-899a-913d689449ec.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Major",
+      "ChangelogMessages": [
+        "Behavior change: under NestingType.Flat, Map and Parallel per-item/branch results are now round-tripped through the configured ItemSerializer on a fresh (non-replay) success -- deserializing the just-serialized inline payload before it enters the result -- matching what replay reconstructs from that payload. A non-round-tripping ItemSerializer transform is now reflected in the item/branch result on the first execution, not only on replay, eliminating a fresh-vs-replay divergence. As with steps and child contexts, the returned value is a fresh deserialized instance, and overflow (replay-children) results are unaffected (recovered by re-running the body)."
+      ]
+    }
+  ]
+}

.autover/changes/d12219eb-e44b-4e90-8473-0ca63319ad53.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Major",
+      "ChangelogMessages": [
+        "Behavior change: a fresh-success serializer round-trip DESERIALIZE failure is now terminal on the child-context and Flat Map/Parallel paths, matching StepOperation. (Child) ChildContextOperation now emits a CONTEXT FAIL for a non-virtual, non-suppressed child whose round-trip deserialize throws, instead of throwing with no terminal checkpoint (which left the child STARTED and re-ran its side-effecting body on replay). (Flat Map/Parallel) a per-item round-trip deserialize failure now records that unit\u0027s terminal Error inline and still emits the parent SUCCEED, instead of propagating raw with no parent checkpoint (which re-ran every virtual unit body on replay, a poison loop, and bypassed the record-failure contract). Replay reconstructs the terminal failure without re-running or re-deserializing."
+      ]
+    }
+  ]
+}

.autover/changes/f3a1520c-8eca-4ac8-a7c6-27168feaff25.json

@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Major",
+      "ChangelogMessages": [
+        "Behavior change (Map/Parallel, NestingType.Nested default): the parent now inlines each Nested unit\u0027s ORIGINAL child SUCCEED-checkpoint payload (S(body)) verbatim on the parent summary, and reconstructs it on replay with the child\u0027s own operation id, instead of re-serializing the already-round-tripped return value at the parent\u0027s per-unit id. This removes a fresh-vs-replay divergence with a non-round-tripping ItemSerializer (fresh vs replay values now match) and, for a context-aware serializer such as FileSystemSerializer, stops the parent writing a second file that orphaned the child\u0027s file. Round-tripping (symmetric) serializers and the default JSON serializer are unaffected; the Flat path is unchanged."
+      ]
+    }
+  ]
+}

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

@@ -1,6 +1,8 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+using Amazon.Lambda.Core;
+
 namespace Amazon.Lambda.DurableExecution;
 
 /// <summary>
@@ -60,6 +62,15 @@ public TimeSpan HeartbeatTimeout
         }
     }
 
+    /// <summary>
+    /// Optional serializer used to <b>deserialize</b> the callback payload delivered by the
+    /// external system. When <c>null</c> (default), the globally-registered
+    /// <see cref="ILambdaSerializer"/> on <see cref="ILambdaContext.Serializer"/> is used.
+    /// Only the deserialize path is used for callbacks — the SDK never serializes a callback
+    /// result (the external system provides the payload).
+    /// </summary>
+    public ILambdaSerializer? Serializer { get; set; }
+
     private static void ValidateTimeout(TimeSpan value, string paramName)
     {
         // Allow Zero (means "not set"); reject negative; reject sub-second

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

@@ -1,6 +1,8 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+using Amazon.Lambda.Core;
+
 namespace Amazon.Lambda.DurableExecution;
 
 /// <summary>
@@ -54,4 +56,25 @@ public sealed class ChildContextConfig
     /// </para>
     /// </remarks>
     public NestingType NestingType { get; set; } = NestingType.Nested;
+
+    /// <summary>
+    /// Optional serializer for this child context's result payload. When <c>null</c>
+    /// (default), the globally-registered <see cref="ILambdaSerializer"/> on
+    /// <see cref="ILambdaContext.Serializer"/> is used.
+    /// </summary>
+    /// <remarks>
+    /// <b>Size-dependent transform on overflow.</b> On a fresh (non-replay) success the
+    /// child context result is round-tripped through this serializer before it is
+    /// returned, so a serializer that transforms the value (e.g. redaction, canonicalization)
+    /// has that transform reflected in the returned result — but <em>only</em> while the
+    /// serialized payload fits inline in a single checkpoint (≈256&#160;KB). If the result
+    /// is large enough to overflow, the payload is not stored inline; the value is instead
+    /// recovered by re-running the child body on replay, which does <em>not</em> apply the
+    /// round-trip transform. A serializer whose <c>Deserialize</c> is not a pure inverse of
+    /// its <c>Serialize</c> will therefore observe a size-dependent difference in the returned
+    /// value (transform applied for small results, skipped for overflowed ones). Prefer a
+    /// round-tripping serializer, or do not depend on the transform being applied to results
+    /// large enough to overflow the checkpoint.
+    /// </remarks>
+    public ILambdaSerializer? Serializer { get; set; }
 }

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

@@ -84,7 +84,8 @@ private Task<T> RunStep<T>(
         StepConfig? config,
         CancellationToken cancellationToken)
     {
-        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = LambdaSerializerHelper.WithDefaultInner(config?.Serializer ?? defaultSerializer, defaultSerializer);
 
         var operationId = _idGenerator.NextId();
         var op = new StepOperation<T>(
@@ -147,7 +148,8 @@ public Task<TState> WaitForConditionAsync<TState>(
         ArgumentNullException.ThrowIfNull(config);
         ArgumentNullException.ThrowIfNull(config.WaitStrategy);
 
-        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = LambdaSerializerHelper.WithDefaultInner(config.Serializer ?? defaultSerializer, defaultSerializer);
         var operationId = _idGenerator.NextId();
         var op = new WaitForConditionOperation<TState>(
             operationId, name, _idGenerator.ParentId, check, config, serializer, Logger,
@@ -161,7 +163,8 @@ private Task<T> RunChildContext<T>(
         ChildContextConfig? config,
         CancellationToken cancellationToken)
     {
-        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = LambdaSerializerHelper.WithDefaultInner(config?.Serializer ?? defaultSerializer, defaultSerializer);
 
         var operationId = _idGenerator.NextId();
 
@@ -183,7 +186,7 @@ private Task<ICallback<T>> RunCallback<T>(
         CallbackConfig? config,
         CancellationToken cancellationToken)
     {
-        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = config?.Serializer ?? LambdaSerializerHelper.GetRequired(LambdaContext);
 
         var operationId = _idGenerator.NextId();
         var op = new CallbackOperation<T>(
@@ -242,11 +245,12 @@ private Task<IBatchResult<T>> RunParallel<T>(
 
         var effectiveConfig = config ?? new ParallelConfig();
 
-        var serializer = LambdaContext.Serializer
-            ?? throw new InvalidOperationException(
-                "No ILambdaSerializer is registered on ILambdaContext.Serializer. " +
-                "Register a serializer via LambdaBootstrapBuilder.Create(handler, serializer) " +
-                "(or in tests, set TestLambdaContext.Serializer).");
+        // Per-branch result serialization: the config's ItemSerializer if set, else the
+        // globally-registered serializer. This is the only serializer ConcurrentOperation
+        // uses (per-unit child results + inline summary results); the aggregate batch
+        // envelope is a source-generated structure and is unaffected.
+        var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = LambdaSerializerHelper.WithDefaultInner(effectiveConfig.ItemSerializer ?? defaultSerializer, defaultSerializer);
 
         var operationId = _idGenerator.NextId();
         var op = new Internal.ParallelOperation<T>(
@@ -275,7 +279,12 @@ private Task<IBatchResult<TResult>> RunMap<TItem, TResult>(
 
         var effectiveConfig = config ?? new MapConfig<TItem>();
 
-        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        // Per-item result serialization: the config's ItemSerializer if set, else the
+        // globally-registered serializer. This is the only serializer ConcurrentOperation
+        // uses (per-unit child results + inline summary results); the aggregate batch
+        // envelope is a source-generated structure and is unaffected.
+        var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = LambdaSerializerHelper.WithDefaultInner(effectiveConfig.ItemSerializer ?? defaultSerializer, defaultSerializer);
 
         var operationId = _idGenerator.NextId();
         var op = new Internal.MapOperation<TItem, TResult>(
@@ -495,7 +504,7 @@ private Task<TResult> RunInvoke<TPayload, TResult>(
         if (string.IsNullOrWhiteSpace(functionName))
             throw new ArgumentException("Function name must not be empty or whitespace.", nameof(functionName));
 
-        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+        var serializer = config?.Serializer ?? LambdaSerializerHelper.GetRequired(LambdaContext);
 
         cancellationToken.ThrowIfCancellationRequested();
 

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

@@ -0,0 +1,39 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+namespace Amazon.Lambda.DurableExecution;
+
+/// <summary>
+/// Identifying information passed to an <see cref="IDurableResultSerializer"/> when a
+/// durable operation's result is (de)serialized. It lets a context-aware serializer
+/// derive a stable, collision-free external location (for example a file path) for the
+/// value being stored.
+/// </summary>
+/// <remarks>
+/// The two values are stable for the lifetime of a durable execution — they do not
+/// change across the multiple Lambda invocations (replays) of that execution.
+/// </remarks>
+public readonly struct DurableSerializationContext
+{
+    /// <summary>
+    /// Stable identifier of the operation whose result is being (de)serialized (the
+    /// durable operation id; for Map/Parallel units, a per-unit id derived from it).
+    /// Unique within a single durable execution.
+    /// </summary>
+    public string EntityId { get; }
+
+    /// <summary>
+    /// ARN of the durable execution. Used to avoid collisions between the stored
+    /// results of different executions.
+    /// </summary>
+    public string DurableExecutionArn { get; }
+
+    /// <summary>Creates a new <see cref="DurableSerializationContext"/>.</summary>
+    /// <param name="entityId">The per-operation entity id.</param>
+    /// <param name="durableExecutionArn">The durable execution ARN.</param>
+    public DurableSerializationContext(string entityId, string durableExecutionArn)
+    {
+        EntityId = entityId;
+        DurableExecutionArn = durableExecutionArn;
+    }
+}