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

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,12 @@
+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; }
+}

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

@@ -0,0 +1,262 @@
+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;
+
+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 existingRecord = _state.GetOperation(operationId, OperationAction.Succeed);
+        if (existingRecord != null)
+        {
+            _state.MarkVisited(operationId, wasReplayed: true);
+            return DeserializeResult<T>(existingRecord.Result, config?.Serializer, operationId);
+        }
+
+        var failedRecord = _state.GetOperation(operationId, OperationAction.Fail);
+        if (failedRecord != null)
+        {
+            _state.MarkVisited(operationId, wasReplayed: true);
+            throw CreateStepException(failedRecord);
+        }
+
+        _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 OperationRecord
+            {
+                OperationId = operationId,
+                Type = OperationType.Step,
+                Action = OperationAction.Succeed,
+                Status = OperationStatus.Succeeded,
+                Name = name,
+                Result = serializedResult,
+                Timestamp = DateTimeOffset.UtcNow.ToString("O")
+            });
+
+            return result;
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+        catch (Exception ex)
+        {
+            _state.AddPendingCheckpoint(new OperationRecord
+            {
+                OperationId = operationId,
+                Type = OperationType.Step,
+                Action = OperationAction.Fail,
+                Status = OperationStatus.Failed,
+                Name = name,
+                Error = ErrorObject.FromException(ex),
+                Timestamp = DateTimeOffset.UtcNow.ToString("O")
+            });
+
+            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 expiresAt = DateTimeOffset.UtcNow.Add(duration);
+
+        var existingRecord = _state.GetOperation(operationId, OperationAction.Start);
+        if (existingRecord?.ExpiresAt != null)
+        {
+            var expiration = DateTimeOffset.Parse(existingRecord.ExpiresAt);
+            if (DateTimeOffset.UtcNow >= expiration)
+            {
+                _state.MarkVisited(operationId, wasReplayed: true);
+                return Task.CompletedTask;
+            }
+        }
+
+        var succeedRecord = _state.GetOperation(operationId, OperationAction.Succeed);
+        if (succeedRecord != null)
+        {
+            _state.MarkVisited(operationId, wasReplayed: true);
+            return Task.CompletedTask;
+        }
+
+        _state.MarkVisited(operationId, wasReplayed: false);
+
+        _state.AddPendingCheckpoint(new OperationRecord
+        {
+            OperationId = operationId,
+            Type = OperationType.Wait,
+            Action = OperationAction.Start,
+            Status = OperationStatus.Pending,
+            Name = name,
+            ExpiresAt = expiresAt.ToString("O"),
+            Timestamp = DateTimeOffset.UtcNow.ToString("O")
+        });
+
+        _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.")]
+    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.")]
+    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(OperationRecord failedRecord)
+    {
+        return new StepException(failedRecord.Error?.ErrorMessage ?? "Step failed")
+        {
+            ErrorType = failedRecord.Error?.ErrorType,
+            ErrorData = failedRecord.Error?.ErrorData,
+            OriginalStackTrace = failedRecord.Error?.StackTrace
+        };
+    }
+
+    #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; }
+}