Add durable execution Step + Wait end-to-end

Implements the minimum viable slice of the Amazon.Lambda.DurableExecution
SDK: a workflow can run StepAsync and WaitAsync against a real Lambda,
with replay-aware checkpointing wired through to the AWS service.

Public API surface introduced:
- DurableFunction.WrapAsync — entry point that handles the durable
  execution envelope (input hydration, output construction, status mapping)
- IDurableContext.StepAsync / WaitAsync (4 Step overloads, 1 Wait)
- StepConfig with retry strategy, semantics, and serializer hooks
- IRetryStrategy + ExponentialRetryStrategy + retry decision factories
- ICheckpointSerializer + DefaultJsonCheckpointSerializer
- [DurableExecution] attribute (recognized by future source generator)
- DurableExecutionException base + StepException

Internals:
- DurableExecutionHandler — Task.WhenAny race between user code and
  the suspension signal, returning Succeeded/Failed/Pending
- ExecutionState — replay-aware operation lookup and pending checkpoint
  buffer
- OperationIdGenerator — deterministic, replay-stable IDs
- TerminationManager — TaskCompletionSource-based suspension trigger
- LambdaDurableServiceClient — wraps AWSSDK.Lambda's checkpoint and
  state APIs

Tests:
- 86 unit tests covering enums, exceptions, models, configs, retry,
  ID generation, termination, execution state, the handler race,
  the context (Step + Wait paths), and the WrapAsync entry point
- 8 end-to-end integration tests deploying real Lambdas via Docker on
  the provided.al2023 runtime: StepWaitStep, MultipleSteps, WaitOnly,
  LongerWait, ReplayDeterminism, RetrySucceeds, RetryExhausts, StepFails

Out of scope (follow-up PRs):
- Callbacks, InvokeAsync, ParallelAsync, MapAsync, RunInChildContextAsync,
  WaitForConditionAsync — interface intentionally does not declare them
- DurableLogger replay-suppression
- Annotations source-generator integration
- DurableTestRunner / Amazon.Lambda.DurableExecution.Testing package
- dotnet new lambda.DurableFunction blueprint

stack-info: PR: #2360, branch: GarrettBeatty/stack/2

Author: GarrettBeatty

Libraries/src/Amazon.Lambda.DurableExecution/Amazon.Lambda.DurableExecution.csproj

@@ -14,6 +14,8 @@
     <EnableTrimAnalyzer>true</EnableTrimAnalyzer>
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
+    <WarningsAsErrors>IL2026,IL2067,IL2075,IL3050</WarningsAsErrors>
   </PropertyGroup>
 
   <ItemGroup>

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

@@ -0,0 +1,19 @@
+namespace Amazon.Lambda.DurableExecution;
+
+/// <summary>
+/// Configuration for step execution.
+/// </summary>
+public class StepConfig
+{
+    /// <summary>
+    /// Custom serializer for the step result. Default is System.Text.Json.
+    /// </summary>
+    public ICheckpointSerializer? Serializer { get; set; }
+
+    // TODO: Retry support is deferred to a follow-up PR. When added, this is
+    // where RetryStrategy and Semantics (AtLeastOncePerRetry / AtMostOncePerRetry)
+    // will live. The follow-up needs to use service-mediated retries (checkpoint
+    // a RETRY operation + suspend the Lambda) rather than an in-process Task.Delay
+    // loop, to match the Python and JS reference SDKs and to avoid billing Lambda
+    // compute time during retry backoff.
+}

Libraries/src/Amazon.Lambda.DurableExecution/Config/StepConfig.cs

@@ -0,0 +1,273 @@
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json;
+using Amazon.Lambda.Core;
+using Amazon.Lambda.DurableExecution.Internal;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate;
+using SdkErrorObject = Amazon.Lambda.Model.ErrorObject;
+using SdkWaitOptions = Amazon.Lambda.Model.WaitOptions;
+
+namespace Amazon.Lambda.DurableExecution;
+
+/// <summary>
+/// Core implementation of IDurableContext providing replay-aware step
+/// execution and waits.
+/// </summary>
+internal sealed class DurableContext : IDurableContext
+{
+    private readonly ExecutionState _state;
+    private readonly TerminationManager _terminationManager;
+    private readonly OperationIdGenerator _idGenerator;
+    private readonly string _durableExecutionArn;
+
+    public DurableContext(
+        ExecutionState state,
+        TerminationManager terminationManager,
+        OperationIdGenerator idGenerator,
+        string durableExecutionArn,
+        ILambdaContext lambdaContext)
+    {
+        _state = state;
+        _terminationManager = terminationManager;
+        _idGenerator = idGenerator;
+        _durableExecutionArn = durableExecutionArn;
+        LambdaContext = lambdaContext;
+    }
+
+    public ILogger Logger => NullLogger.Instance;
+    public IExecutionContext ExecutionContext => new DurableExecutionContext(_durableExecutionArn);
+    public ILambdaContext LambdaContext { get; }
+
+    #region StepAsync
+
+    public async Task<T> StepAsync<T>(
+        Func<IStepContext, Task<T>> func,
+        string? name = null,
+        StepConfig? config = null,
+        CancellationToken cancellationToken = default)
+    {
+        var operationId = _idGenerator.NextId(name);
+
+        var existingOp = _state.GetOperation(operationId, OperationAction.Succeed);
+        if (existingOp != null)
+        {
+            _state.MarkVisited(operationId, wasReplayed: true);
+            return DeserializeResult<T>(existingOp.StepDetails?.Result, config?.Serializer, operationId);
+        }
+
+        var failedOp = _state.GetOperation(operationId, OperationAction.Fail);
+        if (failedOp != null)
+        {
+            _state.MarkVisited(operationId, wasReplayed: true);
+            throw CreateStepException(failedOp);
+        }
+
+        _state.MarkVisited(operationId, wasReplayed: false);
+
+        cancellationToken.ThrowIfCancellationRequested();
+
+        try
+        {
+            var stepContext = new StepContext(operationId, attemptNumber: 1, Logger);
+            var result = await func(stepContext);
+
+            var serializedResult = SerializeResult(result, config?.Serializer, operationId);
+            _state.AddPendingCheckpoint(new SdkOperationUpdate
+            {
+                Id = operationId,
+                Type = OperationTypes.Step,
+                Action = "SUCCEED",
+                SubType = "Step",
+                Name = name,
+                Payload = serializedResult
+            });
+
+            return result;
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+        catch (Exception ex)
+        {
+            _state.AddPendingCheckpoint(new SdkOperationUpdate
+            {
+                Id = operationId,
+                Type = OperationTypes.Step,
+                Action = "FAIL",
+                SubType = "Step",
+                Name = name,
+                Error = ToSdkError(ex)
+            });
+
+            throw new StepException(ex.Message, ex)
+            {
+                ErrorType = ex.GetType().FullName,
+                ErrorData = ex.Message
+            };
+        }
+    }
+
+    public async Task StepAsync(
+        Func<IStepContext, Task> func,
+        string? name = null,
+        StepConfig? config = null,
+        CancellationToken cancellationToken = default)
+    {
+        await StepAsync<object?>(
+            async (ctx) => { await func(ctx); return null; },
+            name, config, cancellationToken);
+    }
+
+    public async Task<T> StepAsync<T>(
+        Func<Task<T>> func,
+        string? name = null,
+        StepConfig? config = null,
+        CancellationToken cancellationToken = default)
+    {
+        return await StepAsync<T>(
+            async (_) => await func(),
+            name, config, cancellationToken);
+    }
+
+    public async Task StepAsync(
+        Func<Task> func,
+        string? name = null,
+        StepConfig? config = null,
+        CancellationToken cancellationToken = default)
+    {
+        await StepAsync<object?>(
+            async (_) => { await func(); return null; },
+            name, config, cancellationToken);
+    }
+
+    #endregion
+
+    #region WaitAsync
+
+    public Task WaitAsync(
+        TimeSpan duration,
+        string? name = null,
+        CancellationToken cancellationToken = default)
+    {
+        if (duration < TimeSpan.FromSeconds(1))
+            throw new ArgumentOutOfRangeException(nameof(duration), "Wait duration must be at least 1 second.");
+
+        cancellationToken.ThrowIfCancellationRequested();
+
+        var operationId = _idGenerator.NextId(name);
+        var waitSeconds = (int)Math.Max(1, Math.Ceiling(duration.TotalSeconds));
+
+        var startedOp = _state.GetOperation(operationId, OperationAction.Start);
+        if (startedOp?.WaitDetails?.ScheduledEndTimestamp is { } expiresAtMs)
+        {
+            if (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() >= expiresAtMs)
+            {
+                _state.MarkVisited(operationId, wasReplayed: true);
+                return Task.CompletedTask;
+            }
+        }
+
+        var succeededOp = _state.GetOperation(operationId, OperationAction.Succeed);
+        if (succeededOp != null)
+        {
+            _state.MarkVisited(operationId, wasReplayed: true);
+            return Task.CompletedTask;
+        }
+
+        _state.MarkVisited(operationId, wasReplayed: false);
+
+        _state.AddPendingCheckpoint(new SdkOperationUpdate
+        {
+            Id = operationId,
+            Type = OperationTypes.Wait,
+            Action = "START",
+            SubType = "Wait",
+            Name = name,
+            WaitOptions = new SdkWaitOptions { WaitSeconds = waitSeconds }
+        });
+
+        _terminationManager.Terminate(TerminationReason.WaitScheduled, $"wait:{name ?? operationId}");
+
+        return new TaskCompletionSource<object?>().Task;
+    }
+
+    #endregion
+
+    #region Helpers
+
+    [UnconditionalSuppressMessage("Trimming", "IL2026",
+        Justification = "Default reflection-based path used in JIT mode. AOT users supply a trim-safe ICheckpointSerializer<T> via StepConfig.Serializer to avoid this code path.")]
+    [UnconditionalSuppressMessage("AOT", "IL3050",
+        Justification = "Default reflection-based path used in JIT mode. AOT users supply a trim-safe ICheckpointSerializer<T> via StepConfig.Serializer to avoid this code path.")]
+    private T DeserializeResult<T>(string? serialized, ICheckpointSerializer? customSerializer, string operationId)
+    {
+        if (serialized == null)
+            return default!;
+
+        if (customSerializer is ICheckpointSerializer<T> typed)
+            return typed.Deserialize(serialized, new SerializationContext(operationId, _durableExecutionArn));
+
+        return JsonSerializer.Deserialize<T>(serialized)!;
+    }
+
+    [UnconditionalSuppressMessage("Trimming", "IL2026",
+        Justification = "Default reflection-based path used in JIT mode. AOT users supply a trim-safe ICheckpointSerializer<T> via StepConfig.Serializer to avoid this code path.")]
+    [UnconditionalSuppressMessage("AOT", "IL3050",
+        Justification = "Default reflection-based path used in JIT mode. AOT users supply a trim-safe ICheckpointSerializer<T> via StepConfig.Serializer to avoid this code path.")]
+    private string SerializeResult<T>(T value, ICheckpointSerializer? customSerializer, string operationId)
+    {
+        if (value == null)
+            return "null";
+
+        if (customSerializer is ICheckpointSerializer<T> typed)
+            return typed.Serialize(value, new SerializationContext(operationId, _durableExecutionArn));
+
+        return JsonSerializer.Serialize(value);
+    }
+
+    private static StepException CreateStepException(Operation failedOp)
+    {
+        var err = failedOp.StepDetails?.Error;
+        return new StepException(err?.ErrorMessage ?? "Step failed")
+        {
+            ErrorType = err?.ErrorType,
+            ErrorData = err?.ErrorData,
+            OriginalStackTrace = err?.StackTrace
+        };
+    }
+
+    private static SdkErrorObject ToSdkError(Exception ex) => new()
+    {
+        ErrorType = ex.GetType().FullName,
+        ErrorMessage = ex.Message,
+        StackTrace = ex.StackTrace?.Split('\n', StringSplitOptions.RemoveEmptyEntries).ToList()
+    };
+
+    #endregion
+}
+
+internal sealed class DurableExecutionContext : IExecutionContext
+{
+    public DurableExecutionContext(string durableExecutionArn)
+    {
+        DurableExecutionArn = durableExecutionArn;
+    }
+
+    public string DurableExecutionArn { get; }
+}
+
+internal sealed class StepContext : IStepContext
+{
+    public StepContext(string operationId, int attemptNumber, ILogger logger)
+    {
+        OperationId = operationId;
+        AttemptNumber = attemptNumber;
+        Logger = logger;
+    }
+
+    public ILogger Logger { get; }
+    public int AttemptNumber { get; }
+    public string OperationId { get; }
+}

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

@@ -0,0 +1,26 @@
+namespace Amazon.Lambda.DurableExecution;
+
+/// <summary>
+/// Marks a Lambda function as a durable execution handler. When used with
+/// [LambdaFunction] from Amazon.Lambda.Annotations, the source generator
+/// produces the handler wrapper and CloudFormation DurableConfig automatically.
+/// </summary>
+[AttributeUsage(AttributeTargets.Method, AllowMultiple = false)]
+public sealed class DurableExecutionAttribute : Attribute
+{
+    /// <summary>
+    /// Maximum execution timeout in seconds. Default is service-determined.
+    /// </summary>
+    public int ExecutionTimeout { get; set; }
+
+    /// <summary>
+    /// Number of days to retain execution history. Default is service-determined.
+    /// </summary>
+    public int RetentionPeriodInDays { get; set; }
+
+    /// <summary>
+    /// Name of a field or property on the class that provides a custom IAmazonLambda client.
+    /// When not specified, the generated code creates a default AmazonLambdaClient.
+    /// </summary>
+    public string? LambdaClientFactory { get; set; }
+}