Track replay state per operation rather than via a global flag

Match the Python / Java / JavaScript reference SDKs' replay-mode model:
the workflow is "replaying" iff it has not yet revisited every
checkpointed completed user-replayable operation. A single global flag
flipped on the first fresh op (the prior model) misclassified workflow-
body code that runs before the first step and would not generalize to
Map/Parallel/Callback later.

ExecutionState changes:
- Replace `Mode`/`ExecutionMode`/`EnterExecutionMode()` with `IsReplaying`
  + `TrackReplay(operationId)`.
- Initial replay decision: any non-EXECUTION op present means we're
  replaying. The service always sends an EXECUTION-type op carrying the
  input payload — that's bookkeeping, not user history, so it does not
  count toward replay (matches Python execution.py:258, Java
  ExecutionManager:81, JS execution-context.ts:62).
- TrackReplay flips IsReplaying false once every checkpointed terminal-
  status non-EXECUTION op has been visited. Terminal set matches
  Python's: SUCCEEDED, FAILED, CANCELLED, STOPPED.

Operation changes:
- DurableOperation.ExecuteAsync calls TrackReplay(OperationId) at the
  top, so every operation participates in visit accounting without each
  subclass needing to remember.
- StepOperation/WaitOperation drop their manual EnterExecutionMode calls.

Tests:
- ExecutionStateTests rewritten around IsReplaying/TrackReplay, including
  pinning regressions: only-EXECUTION-op ⇒ NotReplaying, all-visited ⇒
  flips out of replay, PENDING ops do not block transition, idempotency.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: GarrettBeatty

Libraries/src/Amazon.Lambda.DurableExecution/Internal/DurableOperation.cs

@@ -44,6 +44,10 @@ public Task<TResult> ExecuteAsync(CancellationToken cancellationToken)
     {
         State.ValidateReplayConsistency(OperationId, OperationType, Name);
 
+        // Record that the workflow has reached this op. If every completed
+        // checkpointed op has now been visited, the state flips out of replay.
+        State.TrackReplay(OperationId);
+
         var existing = State.GetOperation(OperationId);
         return existing == null
             ? StartAsync(cancellationToken)

Libraries/src/Amazon.Lambda.DurableExecution/Internal/ExecutionState.cs

@@ -1,43 +1,50 @@
 namespace Amazon.Lambda.DurableExecution.Internal;
 
 /// <summary>
-/// Replay state of the current invocation.
-/// </summary>
-internal enum ExecutionMode
-{
-    /// <summary>Re-deriving prior operations from checkpointed state.</summary>
-    Replay,
-    /// <summary>Executing fresh code that hasn't been checkpointed before.</summary>
-    Execution
-}
-
-/// <summary>
-/// In-memory store of the operations replayed from <see cref="InitialExecutionState"/>.
-/// Read-only after load (apart from <see cref="EnterExecutionMode"/>); outbound
-/// checkpoints are owned by <see cref="CheckpointBatcher"/>.
+/// In-memory store of the operations replayed from <see cref="InitialExecutionState"/>
+/// plus replay-mode tracking. Outbound checkpoints are owned by
+/// <see cref="CheckpointBatcher"/>; this type is the inbound side only.
 /// </summary>
+/// <remarks>
+/// Replay tracking mirrors the Python / Java / JavaScript reference SDKs:
+/// <list type="bullet">
+///   <item>At construction the workflow is "replaying" iff any user-replayable
+///       op is present. The service always sends one <c>EXECUTION</c>-type op
+///       carrying the input payload — that's bookkeeping, not user history,
+///       so it doesn't count.</item>
+///   <item><see cref="TrackReplay"/> is called by every <c>DurableOperation.ExecuteAsync</c>
+///       at the top of the call. Once every checkpointed completed
+///       non-<c>EXECUTION</c> op has been visited, the workflow has caught up
+///       to the replay frontier and <see cref="IsReplaying"/> flips to <c>false</c>
+///       for the rest of the invocation.</item>
+/// </list>
+/// </remarks>
 internal sealed class ExecutionState
 {
     private readonly Dictionary<string, Operation> _operations = new();
-
-    public ExecutionMode Mode { get; private set; } = ExecutionMode.Replay;
+    private readonly HashSet<string> _visitedOperations = new();
+    private bool _isReplaying;
 
     public int CheckpointedOperationCount => _operations.Count;
 
+    /// <summary>
+    /// True when the workflow is re-deriving prior operations from checkpointed
+    /// state. False when running fresh (not-yet-checkpointed) code.
+    /// </summary>
+    public bool IsReplaying => _isReplaying;
+
     public void LoadFromCheckpoint(InitialExecutionState? initialState)
     {
-        if (initialState?.Operations == null)
+        if (initialState?.Operations != null)
         {
-            Mode = ExecutionMode.Execution;
-            return;
+            AddOperations(initialState.Operations);
         }
 
-        AddOperations(initialState.Operations);
-
-        if (_operations.Count == 0)
-        {
-            Mode = ExecutionMode.Execution;
-        }
+        // Only user-replayable ops put us into replay mode. The service-side
+        // EXECUTION op (input payload bookkeeping) is always present and must
+        // not count — see Python execution.py:258 / Java ExecutionManager:81 /
+        // JS execution-context.ts:62 for the same rule.
+        _isReplaying = HasReplayableOperations();
     }
 
     public void AddOperations(IEnumerable<Operation> operations)
@@ -60,9 +67,36 @@ public void AddOperations(IEnumerable<Operation> operations)
         return op;
     }
 
+    public bool HasOperation(string operationId) => _operations.ContainsKey(operationId);
+
+    /// <summary>
+    /// Records that the workflow has reached <paramref name="operationId"/>.
+    /// Once every checkpointed completed non-<c>EXECUTION</c> op has been
+    /// visited the workflow has caught up to the replay frontier and
+    /// <see cref="IsReplaying"/> flips to false. Idempotent: calling more than
+    /// once with the same id has no additional effect.
+    /// </summary>
+    public void TrackReplay(string operationId)
+    {
+        if (!_isReplaying) return;
+
+        _visitedOperations.Add(operationId);
+
+        // Have we visited every completed non-EXECUTION op? If so, anything
+        // emitted from here on is fresh execution.
+        foreach (var op in _operations.Values)
+        {
+            if (op.Type == OperationTypes.Execution) continue;
+            if (!IsTerminalStatus(op.Status)) continue;
+            if (!_visitedOperations.Contains(op.Id!)) return;
+        }
+
+        _isReplaying = false;
+    }
+
     public void ValidateReplayConsistency(string operationId, string expectedType, string? expectedName)
     {
-        if (Mode != ExecutionMode.Replay) return;
+        if (!_isReplaying) return;
 
         if (!_operations.TryGetValue(operationId, out var op)) return;
 
@@ -83,11 +117,18 @@ public void ValidateReplayConsistency(string operationId, string expectedType, s
         }
     }
 
-    public bool HasOperation(string operationId) => _operations.ContainsKey(operationId);
+    private bool HasReplayableOperations()
+    {
+        foreach (var op in _operations.Values)
+        {
+            if (op.Type != OperationTypes.Execution) return true;
+        }
+        return false;
+    }
 
-    /// <summary>
-    /// Transitions to <see cref="ExecutionMode.Execution"/>. Called by an operation
-    /// that's about to run fresh (not-yet-checkpointed) code. Idempotent.
-    /// </summary>
-    public void EnterExecutionMode() => Mode = ExecutionMode.Execution;
+    private static bool IsTerminalStatus(string? status) =>
+        status == OperationStatuses.Succeeded
+        || status == OperationStatuses.Failed
+        || status == OperationStatuses.Cancelled
+        || status == OperationStatuses.Stopped;
 }

Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs

@@ -50,10 +50,7 @@ public StepOperation(
     protected override string OperationType => OperationTypes.Step;
 
     protected override Task<T> StartAsync(CancellationToken cancellationToken)
-    {
-        State.EnterExecutionMode();
-        return ExecuteFunc(cancellationToken);
-    }
+        => ExecuteFunc(cancellationToken);
 
     protected override Task<T> ReplayAsync(Operation existing, CancellationToken cancellationToken)
     {
@@ -73,7 +70,6 @@ protected override Task<T> ReplayAsync(Operation existing, CancellationToken can
                 // STARTED/READY/PENDING from a prior invocation — no retry logic
                 // in this commit, so fall through and execute fresh. (Future work
                 // on retries will replace this default with explicit arms.)
-                State.EnterExecutionMode();
                 return ExecuteFunc(cancellationToken);
         }
     }

Libraries/src/Amazon.Lambda.DurableExecution/Internal/WaitOperation.cs

@@ -41,8 +41,6 @@ public WaitOperation(
 
     protected override async Task<object?> StartAsync(CancellationToken cancellationToken)
     {
-        State.EnterExecutionMode();
-
         // Sync-flush WAIT START before suspending — the service can't schedule
         // a timer for a checkpoint it hasn't received.
         await EnqueueAsync(new SdkOperationUpdate