diff --git a/.autover/changes/46bb3a38-e1f3-48d1-ab9b-10a36475e901.json b/.autover/changes/46bb3a38-e1f3-48d1-ab9b-10a36475e901.json
new file mode 100644
index 000000000..4eee32762
--- /dev/null
+++ b/.autover/changes/46bb3a38-e1f3-48d1-ab9b-10a36475e901.json
@@ -0,0 +1,12 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Patch",
+      "ChangelogMessages": [
+        "Cancel user work that is still running when an invocation ends, and reject new durable operations started after that point.",
+        "On Lambda Managed Instances, stop durable work at the invocation deadline and throw TimeoutException so the invocation is retried instead of left running."
+      ]
+    }
+  ]
+}
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs b/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs
index 11530af7d..d2444c3a8 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs
@@ -84,6 +84,7 @@ private Task<T> RunStep<T>(
         StepConfig? config,
         CancellationToken cancellationToken)
     {
+        _workflowCancellation.ThrowIfTornDown();
         var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
         var serializer = LambdaSerializerHelper.WithDefaultInner(config?.Serializer ?? defaultSerializer, defaultSerializer);
 
@@ -99,6 +100,7 @@ public Task WaitAsync(
         string? name = null,
         CancellationToken cancellationToken = default)
     {
+        _workflowCancellation.ThrowIfTornDown();
         // Service timer granularity is 1 second; sub-second waits would round to 0.
         // WaitOptions.WaitSeconds is integer in [1, 31_622_400] (1 second to ~1 year).
         if (duration < TimeSpan.FromSeconds(1))
@@ -144,6 +146,7 @@ public Task<TState> WaitForConditionAsync<TState>(
         string? name = null,
         CancellationToken cancellationToken = default)
     {
+        _workflowCancellation.ThrowIfTornDown();
         ArgumentNullException.ThrowIfNull(check);
         ArgumentNullException.ThrowIfNull(config);
         ArgumentNullException.ThrowIfNull(config.WaitStrategy);
@@ -163,6 +166,7 @@ private Task<T> RunChildContext<T>(
         ChildContextConfig? config,
         CancellationToken cancellationToken)
     {
+        _workflowCancellation.ThrowIfTornDown();
         var defaultSerializer = LambdaSerializerHelper.GetRequired(LambdaContext);
         var serializer = LambdaSerializerHelper.WithDefaultInner(config?.Serializer ?? defaultSerializer, defaultSerializer);
 
@@ -186,6 +190,7 @@ private Task<ICallback<T>> RunCallback<T>(
         CallbackConfig? config,
         CancellationToken cancellationToken)
     {
+        _workflowCancellation.ThrowIfTornDown();
         var serializer = config?.Serializer ?? LambdaSerializerHelper.GetRequired(LambdaContext);
 
         var operationId = _idGenerator.NextId();
@@ -199,6 +204,7 @@ public IDurableParallel CreateParallel(
         string? name = null,
         ParallelConfig? config = null)
     {
+        _workflowCancellation.ThrowIfTornDown();
         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
@@ -266,6 +272,7 @@ private Task<IBatchResult<T>> RunParallel<T>(
         ParallelConfig? config,
         CancellationToken cancellationToken)
     {
+        _workflowCancellation.ThrowIfTornDown();
         if (branches == null) throw new ArgumentNullException(nameof(branches));
         for (var i = 0; i < branches.Count; i++)
         {
@@ -306,6 +313,7 @@ private Task<IBatchResult<TResult>> RunMap<TItem, TResult>(
         MapConfig<TItem>? config,
         CancellationToken cancellationToken)
     {
+        _workflowCancellation.ThrowIfTornDown();
         if (items == null) throw new ArgumentNullException(nameof(items));
         if (func == null) throw new ArgumentNullException(nameof(func));
 
@@ -528,6 +536,7 @@ private Task<TResult> RunInvoke<TPayload, TResult>(
         InvokeConfig? config,
         CancellationToken cancellationToken)
     {
+        _workflowCancellation.ThrowIfTornDown();
         // Argument validation runs synchronously at the call site (matches the
         // .NET convention of failing fast for misuse). Match Python/JS/Java
         // parity: only check for null/empty here; the durable execution service
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionHandler.cs b/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionHandler.cs
index 971ac6f64..395a61cec 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionHandler.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionHandler.cs
@@ -14,6 +14,13 @@ internal sealed class HandlerResult<TResult>
     public TResult? Result { get; init; }
     public string? Message { get; init; }
     public Exception? Exception { get; init; }
+
+    /// <summary>
+    /// True when the invocation deadline passed before the workflow completed.
+    /// <see cref="Status"/> is meaningless in that case: the wrapper fails the
+    /// invocation so Lambda retries it, rather than recording a durable outcome.
+    /// </summary>
+    public bool DeadlineExceeded { get; init; }
 }
 
 /// <summary>
@@ -23,6 +30,9 @@ internal sealed class HandlerResult<TResult>
 /// </summary>
 internal static class DurableExecutionHandler
 {
+    // Stands in for the deadline when the caller passes no WorkflowCancellation.
+    private static readonly Task NeverCompletes = new TaskCompletionSource().Task;
+
     /// <summary>
     /// Runs the user's workflow function within the durable execution engine.
     /// </summary>
@@ -49,8 +59,10 @@ internal static class DurableExecutionHandler
     /// Key insight: <c>WaitAsync</c> never returns a completed Task — it hands back
     /// a TaskCompletionSource that is never resolved. The user's <c>await</c> blocks
     /// indefinitely. The escape signal is <c>terminationManager.Terminate()</c>,
-    /// which <c>Task.WhenAny</c> picks up. We return Pending; the dangling user
-    /// Task is GC'd. The service flushes checkpoints, fires the wait timer, then
+    /// which <c>Task.WhenAny</c> picks up. We return Pending and observe the
+    /// abandoned user Task's eventual fault; <see cref="DurableFunction"/> then
+    /// cancels the workflow token so any user work still running unwinds. The
+    /// service flushes checkpoints, fires the wait timer, then
     /// re-invokes Lambda — on replay, <c>WaitAsync</c> sees the matching SUCCEED
     /// checkpoint and returns <c>Task.CompletedTask</c> normally.
     /// </para>
@@ -63,21 +75,61 @@ internal static class DurableExecutionHandler
     /// <param name="executionState">Hydrated execution state from prior invocations.</param>
     /// <param name="terminationManager">Manages the suspension signal.</param>
     /// <param name="userHandler">The user's workflow function receiving a DurableContext.</param>
+    /// <param name="workflowCancellation">
+    /// Optional workflow cancellation for this invocation. When it carries a
+    /// deadline (Lambda Managed Instances only), the deadline joins the race.
+    /// </param>
     /// <returns>The handler result indicating SUCCEEDED, FAILED, or PENDING.</returns>
     internal static async Task<HandlerResult<TResult>> RunAsync<TResult>(
         ExecutionState executionState,
         TerminationManager terminationManager,
-        Func<Task<TResult>> userHandler)
+        Func<Task<TResult>> userHandler,
+        WorkflowCancellation? workflowCancellation = null)
     {
         // Run user code on a threadpool thread so it executes independently of
         // the termination signal. When TerminationManager fires (e.g., WaitAsync),
         // we need the WhenAny race below to resolve immediately without waiting
         // for the user task to reach an await point.
-        var userTask = Task.Run(userHandler);
+        //
+        // The workflow is only queued here; under load it may start after the
+        // deadline passed or after this invocation already returned. Check again
+        // when it actually starts so a late start runs no workflow code.
+        var userTask = Task.Run(() =>
+        {
+            workflowCancellation?.ThrowIfTornDown();
+            return userHandler();
+        });
+
+        // Race: user code completing vs. termination signal (wait/callback/retry)
+        // vs. the invocation deadline. DeadlineTask never completes when there is
+        // no deadline, so on-demand Lambda races only the first two.
+        var deadlineTask = workflowCancellation?.DeadlineTask ?? NeverCompletes;
+        var winner = await Task.WhenAny(userTask, terminationManager.TerminationTask, deadlineTask);
 
-        // Race: user code completing vs. termination signal (wait/callback/retry).
-        // If termination wins, we return PENDING and the abandoned userTask is never awaited.
-        var winner = await Task.WhenAny(userTask, terminationManager.TerminationTask);
+        // Once the deadline has passed, it decides the outcome whichever task won.
+        // A workflow can catch the cancellation and return a fallback value, or a
+        // termination can race the deadline; neither is a real result, and the
+        // service must re-invoke the execution.
+        var deadlineExceeded = workflowCancellation?.IsDeadlineExceeded == true;
+
+        if (winner != userTask || deadlineExceeded)
+        {
+            // userTask is abandoned or its outcome is ignored. Observe its fault so
+            // it never surfaces through TaskScheduler.UnobservedTaskException, which
+            // in a shared process (Lambda Managed Instances) would be reported
+            // against whichever invocation happens to be running when the GC
+            // finalizes it.
+            ObserveAbandoned(userTask);
+        }
+
+        if (deadlineExceeded)
+        {
+            return new HandlerResult<TResult>
+            {
+                Status = InvocationStatus.Failed,
+                DeadlineExceeded = true,
+            };
+        }
 
         if (winner == terminationManager.TerminationTask)
         {
@@ -111,6 +163,18 @@ internal static async Task<HandlerResult<TResult>> RunAsync<TResult>(
         }
         catch (Exception ex)
         {
+            // A workflow that faulted because the deadline cancelled it did not
+            // really fail. Report the deadline instead, so the invocation is
+            // retried rather than the execution being failed permanently.
+            if (workflowCancellation?.IsDeadlineExceeded == true)
+            {
+                return new HandlerResult<TResult>
+                {
+                    Status = InvocationStatus.Failed,
+                    DeadlineExceeded = true,
+                };
+            }
+
             return new HandlerResult<TResult>
             {
                 Status = InvocationStatus.Failed,
@@ -119,4 +183,11 @@ internal static async Task<HandlerResult<TResult>> RunAsync<TResult>(
             };
         }
     }
+
+    private static void ObserveAbandoned(Task task) =>
+        task.ContinueWith(
+            static t => _ = t.Exception,
+            CancellationToken.None,
+            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+            TaskScheduler.Default);
 }
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/DurableFunction.cs b/Libraries/src/Amazon.Lambda.DurableExecution/DurableFunction.cs
index 895e96631..d195b18d9 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/DurableFunction.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/DurableFunction.cs
@@ -1,6 +1,7 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+using System.Diagnostics;
 using System.IO;
 using System.Text;
 using System.Threading;
@@ -83,33 +84,76 @@ internal static Task<DurableExecutionInvocationOutput> WrapAsync<TInput, TOutput
         DurableExecutionInvocationInput invocationInput,
         ILambdaContext lambdaContext,
         IDurableServiceClient serviceClient)
-        => WrapAsyncCore(workflow, invocationInput, lambdaContext, serviceClient);
+        => WrapAsyncCore(workflow, invocationInput, lambdaContext, serviceClient, options: null);
+
+    /// <summary>
+    /// Internal overload for tests that override <see cref="DurableInvocationOptions"/>
+    /// (Lambda Managed Instances detection and the cleanup budget).
+    /// </summary>
+    internal static Task<DurableExecutionInvocationOutput> WrapAsync<TInput, TOutput>(
+        Func<TInput, IDurableContext, Task<TOutput>> workflow,
+        DurableExecutionInvocationInput invocationInput,
+        ILambdaContext lambdaContext,
+        IDurableServiceClient serviceClient,
+        DurableInvocationOptions options)
+        => WrapAsyncCore(workflow, invocationInput, lambdaContext, serviceClient, options);
 
     private static async Task<DurableExecutionInvocationOutput> WrapAsyncCore<TInput, TOutput>(
         Func<TInput, IDurableContext, Task<TOutput>> workflow,
         DurableExecutionInvocationInput invocationInput,
         ILambdaContext lambdaContext,
-        IDurableServiceClient serviceClient)
+        IDurableServiceClient serviceClient,
+        DurableInvocationOptions? options = null)
     {
         var serializer = LambdaSerializerHelper.GetRequired(lambdaContext);
 
+        // Lambda Managed Instances only: run the invocation against its deadline.
+        // On-demand Lambda gets no deadline, so every wait below is unbounded as
+        // before and the environment's own timeout stops the invocation.
+        var startTimestamp = Stopwatch.GetTimestamp();
+        var deadline = InvocationDeadline.Resolve(
+            lambdaContext.RemainingTime,
+            options?.IsMultiConcurrency ?? InvocationDeadline.IsMultiConcurrency());
+        var cleanupTimeout = options?.CleanupTimeout ?? DurableInvocationOptions.DefaultCleanupTimeout;
+
+        var terminationManager = new TerminationManager();
+        // Anchor the deadline to when RemainingTime was read, so setup time is not added to it.
+        using var workflowCancellation = new WorkflowCancellation(terminationManager, deadline, startTimestamp);
+        options?.OnWorkflowCancellationCreated?.Invoke(workflowCancellation);
+
         var state = new ExecutionState();
         state.LoadFromCheckpoint(invocationInput.InitialExecutionState);
 
         var checkpointToken = invocationInput.CheckpointToken;
 
+        // Lambda Managed Instances: an invocation can reach the handler with its
+        // deadline already passed (dispatch or deserialization took too long).
+        if (workflowCancellation.IsDeadlineExceeded)
+            throw CreateDeadlineException(invocationInput.DurableExecutionArn, innerException: null);
+
         var nextMarker = invocationInput.InitialExecutionState?.NextMarker;
-        while (!string.IsNullOrEmpty(nextMarker))
+        try
+        {
+            while (!string.IsNullOrEmpty(nextMarker))
+            {
+                var page = serviceClient.GetExecutionStateAsync(
+                    invocationInput.DurableExecutionArn, checkpointToken, nextMarker,
+                    workflowCancellation.DeadlineToken);
+                // Bound the wait too, in case the call does not honor the token.
+                var (operations, marker) = await WaitWithDeadlineAsync(page, workflowCancellation);
+                state.AddOperations(operations);
+                nextMarker = marker;
+            }
+        }
+        catch (Exception ex) when (workflowCancellation.IsDeadlineExceeded)
         {
-            var (operations, marker) = await serviceClient.GetExecutionStateAsync(
-                invocationInput.DurableExecutionArn, checkpointToken, nextMarker);
-            state.AddOperations(operations);
-            nextMarker = marker;
+            // Any hydration failure once the deadline has passed (a cancelled
+            // request, or one that faulted as it was cut off) is reported as the
+            // timeout, with the original kept as the inner exception.
+            throw CreateDeadlineException(invocationInput.DurableExecutionArn, ex);
         }
 
         var userPayload = ExtractUserPayload<TInput>(invocationInput, serializer);
-        var terminationManager = new TerminationManager();
-        using var workflowCancellation = new WorkflowCancellation(terminationManager);
         var idGenerator = new OperationIdGenerator();
 
         await using var batcher = new CheckpointBatcher(
@@ -145,15 +189,45 @@ private static async Task<DurableExecutionInvocationOutput> WrapAsyncCore<TInput
             {
                 result = await DurableExecutionHandler.RunAsync<TOutput>(
                     state, terminationManager,
-                    async () => await workflow(userPayload, context));
+                    async () => await workflow(userPayload, context),
+                    workflowCancellation);
+            }
+
+            // The invocation's outcome is decided. Cancel the workflow token now,
+            // synchronously, so user work abandoned by a suspension (a losing
+            // parallel branch, a step body still running) unwinds before the
+            // drain below. Waiting for TerminationManager's threadpool
+            // continuation instead could let it run after disposal, losing the
+            // cancel. From here on, new durable operations are rejected.
+            workflowCancellation.BeginTeardown();
+
+            if (deadline is null)
+            {
+                await batcher.DrainAsync();
+            }
+            else
+            {
+                // Lambda Managed Instances: a stalled checkpoint call must not hold
+                // this invocation (and one of the runtime's processing loops) open
+                // indefinitely. Allow time up to the deadline plus a fixed cleanup
+                // budget, then abandon the flush.
+                var drainTimeout = ComputeDrainTimeout(deadline.Value, cleanupTimeout, Stopwatch.GetElapsedTime(startTimestamp));
+                if (!await batcher.DrainAsync(drainTimeout))
+                    throw CreateDeadlineException(invocationInput.DurableExecutionArn, innerException: null);
             }
 
-            await batcher.DrainAsync();
+            // Check the deadline itself, not just the handler's result: it can
+            // pass during the drain, after the result was decided.
+            if (result.DeadlineExceeded || workflowCancellation.IsDeadlineExceeded)
+                throw CreateDeadlineException(invocationInput.DurableExecutionArn, innerException: null);
 
             if (batcher.IsCheckpointTokenMissing)
                 result = new HandlerResult<TOutput> { Status = InvocationStatus.Pending };
         }
-        catch (DurableExecutionException ex) when (ex.InnerException is AmazonServiceException sdkEx && IsTerminalCheckpointError(sdkEx))
+        catch (DurableExecutionException ex) when (
+            !workflowCancellation.IsDeadlineExceeded
+            && ex.InnerException is AmazonServiceException sdkEx
+            && IsTerminalCheckpointError(sdkEx))
         {
             return new DurableExecutionInvocationOutput
             {
@@ -161,10 +235,85 @@ private static async Task<DurableExecutionInvocationOutput> WrapAsyncCore<TInput
                 Error = ErrorObject.FromException(ex)
             };
         }
+        catch (Exception ex) when (workflowCancellation.IsDeadlineExceeded && ex is not TimeoutException)
+        {
+            // Once the deadline has passed, any failure (a cancelled checkpoint
+            // call, a checkpoint rejected for the stale token) is a consequence of
+            // the timeout. Report the timeout so the invocation is retried, never
+            // a permanent durable failure.
+            throw CreateDeadlineException(invocationInput.DurableExecutionArn, ex);
+        }
+
+        // Serializing the output runs user serializer code, which can take long
+        // enough to cross the deadline. Report that as a timeout too; a serializer
+        // failure after the deadline is kept as the inner exception.
+        DurableExecutionInvocationOutput output;
+        try
+        {
+            output = MapToOutput(result, serializer);
+        }
+        catch (Exception ex) when (workflowCancellation.IsDeadlineExceeded)
+        {
+            throw CreateDeadlineException(invocationInput.DurableExecutionArn, ex);
+        }
+
+        if (workflowCancellation.IsDeadlineExceeded)
+            throw CreateDeadlineException(invocationInput.DurableExecutionArn, innerException: null);
+
+        return output;
+    }
+
+    /// <summary>
+    /// How long the final checkpoint drain may run on Lambda Managed Instances:
+    /// what is left of one combined budget (the deadline plus the cleanup
+    /// budget), measured from invocation start. Time already spent past the
+    /// deadline comes out of the cleanup budget rather than restarting it.
+    /// </summary>
+    internal static TimeSpan ComputeDrainTimeout(TimeSpan deadline, TimeSpan cleanupTimeout, TimeSpan elapsed)
+    {
+        var remaining = deadline + cleanupTimeout - elapsed;
+        return remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero;
+    }
+
+    /// <summary>
+    /// Awaits <paramref name="task"/>, but stops waiting when the invocation
+    /// deadline passes, and then throws <see cref="OperationCanceledException"/>.
+    /// Races <see cref="WorkflowCancellation.DeadlineTask"/>, which the SDK
+    /// signals before any foreign callback runs, rather than the deadline token:
+    /// a blocking callback the client registered on that token could otherwise
+    /// delay the wait. A task that is abandoned this way has its eventual fault
+    /// observed. With no deadline this is a plain await.
+    /// </summary>
+    internal static async Task<T> WaitWithDeadlineAsync<T>(Task<T> task, WorkflowCancellation workflowCancellation)
+    {
+        if (!workflowCancellation.DeadlineToken.CanBeCanceled)
+            return await task.ConfigureAwait(false);
+
+        if (await Task.WhenAny(task, workflowCancellation.DeadlineTask).ConfigureAwait(false) == task)
+            return await task.ConfigureAwait(false);
 
-        return MapToOutput(result, serializer);
+        // Nothing else ever awaits the abandoned request; observe its fault.
+        // A request that honored the token completes cancelled and needs nothing.
+        _ = task.ContinueWith(
+            static t => _ = t.Exception,
+            CancellationToken.None,
+            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+            TaskScheduler.Default);
+        throw new OperationCanceledException("The invocation deadline passed while loading execution state.",
+            workflowCancellation.DeadlineToken);
     }
 
+    /// <summary>
+    /// The exception thrown when a Lambda Managed Instances invocation runs out
+    /// of time. It escapes to the Lambda runtime as an invocation error, the same
+    /// outcome as an on-demand timeout, so the durable execution is re-invoked and
+    /// replays from its last checkpoint instead of being marked failed.
+    /// </summary>
+    private static TimeoutException CreateDeadlineException(string durableExecutionArn, Exception? innerException) =>
+        new($"The invocation for durable execution '{durableExecutionArn}' reached its deadline before it completed. " +
+            "Durable work for this invocation was cancelled; work that was not checkpointed runs again on replay.",
+            innerException);
+
     /// <summary>
     /// Returns true for checkpoint-flush SDK errors that should fail the workflow
     /// (Failed envelope) instead of escaping to the host (Lambda retry). The catch
@@ -187,7 +336,7 @@ private static async Task<DurableExecutionInvocationOutput> WrapAsyncCore<TInput
     ///   1. A flush triggered synchronously from inside a user <c>StepAsync</c> call
     ///      (the user awaits <c>EnqueueAsync</c> → batch flush → SDK throws → service client
     ///      wraps).
-    ///   2. The final <see cref="CheckpointBatcher.DrainAsync"/> after the workflow returns.
+    ///   2. The final <see cref="CheckpointBatcher.DrainAsync()"/> after the workflow returns.
     ///
     /// State-hydration errors (<c>GetExecutionStateAsync</c>) propagate as
     /// <see cref="DurableExecutionException"/> too, but they are NOT caught here — they
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CheckpointBatcher.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CheckpointBatcher.cs
index 588263096..90c8c0cbe 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CheckpointBatcher.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CheckpointBatcher.cs
@@ -16,7 +16,7 @@ namespace Amazon.Lambda.DurableExecution.Internal;
 /// <remarks>
 /// Fire-and-forget semantics are achieved by simply not awaiting the returned
 /// Task. Errors still surface deterministically via <c>_terminalError</c>: the
-/// next sync <see cref="EnqueueAsync"/> or <see cref="DrainAsync"/> rethrows.
+/// next sync <see cref="EnqueueAsync"/> or <see cref="DrainAsync()"/> rethrows.
 /// Callers using fire-and-forget should observe the discarded Task's exception
 /// (see <c>StepOperation.FireAndForget</c>) so it doesn't trip the runtime's
 /// <c>UnobservedTaskException</c> event.
@@ -31,9 +31,15 @@ internal sealed class CheckpointBatcher : IAsyncDisposable
     private readonly Task _worker;
     private readonly CancellationTokenSource _shutdownCts = new();
 
+    // Cancelled only when the worker is abandoned (Lambda Managed Instances
+    // cleanup budget exceeded). Never disposed: it is cheap, and abandoned
+    // awaiters may still be registered on it.
+    private readonly CancellationTokenSource _abandonCts = new();
+
     private string? _checkpointToken;
     private Exception? _terminalError;
     private int _disposed;
+    private int _abandoned;
     private int _tokenMissing;
 
     // Per-update wire-footprint estimate constants. Deliberate over-estimates:
@@ -116,9 +122,32 @@ public async Task EnqueueAsync(SdkOperationUpdate update, CancellationToken canc
         }
 
         // After a missing token this never completes (the invocation is suspended).
-        await tcs.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+        // It is also released when the worker is abandoned: a stalled flush may
+        // never return, and the waiting workflow must still be able to unwind
+        // rather than being kept alive by that request.
+        var completion = tcs.Task.WaitAsync(cancellationToken);
+        try
+        {
+            await completion.WaitAsync(_abandonCts.Token).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (_abandonCts.IsCancellationRequested)
+        {
+            // The worker may fail this item later, or may have just failed it after
+            // the abandonment won the wait. Either way nothing else awaits it, so
+            // always observe it; it never reaches TaskScheduler.UnobservedTaskException.
+            ObserveFault(completion);
+            throw new OperationCanceledException(
+                "The checkpoint was not confirmed before the invocation's cleanup budget ran out.", _abandonCts.Token);
+        }
     }
 
+    private static void ObserveFault(Task task) =>
+        task.ContinueWith(
+            static t => _ = t.Exception,
+            CancellationToken.None,
+            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+            TaskScheduler.Default);
+
     /// <summary>
     /// Closes the channel and awaits the worker. Any items already enqueued are
     /// flushed; any subsequent <see cref="EnqueueAsync"/> call throws.
@@ -139,11 +168,66 @@ public async Task DrainAsync()
         if (terminal != null) ExceptionDispatchInfo.Throw(terminal);
     }
 
+    /// <summary>
+    /// Like <see cref="DrainAsync()"/>, but gives up after <paramref name="timeout"/>.
+    /// Used for Lambda Managed Instances, where a stalled checkpoint call must not
+    /// hold the invocation open past its deadline.
+    /// </summary>
+    /// <returns>
+    /// <c>true</c> when the worker finished. <c>false</c> when the timeout elapsed:
+    /// the in-flight flush is cancelled and the worker is abandoned, so any
+    /// checkpoint it had not yet sent is not persisted (replay re-runs that work).
+    /// </returns>
+    public async Task<bool> DrainAsync(TimeSpan timeout)
+    {
+        _channel.Writer.TryComplete();
+        try
+        {
+            await _worker.WaitAsync(timeout).ConfigureAwait(false);
+        }
+        catch (TimeoutException)
+        {
+            Abandon();
+            return false;
+        }
+        catch
+        {
+            // Surfaced via _terminalError below.
+        }
+
+        var terminal = Volatile.Read(ref _terminalError);
+        if (terminal != null) ExceptionDispatchInfo.Throw(terminal);
+        return true;
+    }
+
+    private void Abandon()
+    {
+        if (Interlocked.Exchange(ref _abandoned, 1) != 0) return;
+
+        // Release every EnqueueAsync awaiter, in flight or queued. Only the SDK's
+        // own WaitAsync promises are registered on this token, so cancelling it
+        // synchronously runs no foreign code.
+        _abandonCts.Cancel();
+
+        // Ask the in-flight flush to stop. CancelAsync runs the callbacks the
+        // service client registered on the threadpool, so a callback that blocks
+        // or throws cannot stall the drain past its budget. A flush that ignores
+        // the token keeps running until it returns; its outcome is not awaited.
+        // The source is never disposed once abandoned: the worker may still read it.
+        ObserveFault(_shutdownCts.CancelAsync());
+        ObserveFault(_worker);
+    }
+
     public async ValueTask DisposeAsync()
     {
         if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
 
         _channel.Writer.TryComplete();
+
+        // An abandoned worker may never finish. Don't wait for it, and leave its
+        // cancellation source undisposed because the worker may still read the token.
+        if (Volatile.Read(ref _abandoned) == 1) return;
+
         _shutdownCts.Cancel();
         try { await _worker.ConfigureAwait(false); }
         catch { /* swallow on dispose */ }
@@ -254,6 +338,13 @@ private async Task FlushBatchAsync(IReadOnlyList<BatchItem> batch, CancellationT
 
         try
         {
+            // Never start a new request once the worker is shut down or abandoned.
+            // If an abandoned flush ignored cancellation and later returned, the
+            // worker would otherwise go on to send queued batches with this
+            // invocation's checkpoint token, racing the invocation that replays it.
+            // Throwing here routes through the catch below, which fails the batch.
+            cancellationToken.ThrowIfCancellationRequested();
+
             var newToken = await _flushAsync(_checkpointToken, updates, cancellationToken).ConfigureAwait(false);
             Volatile.Write(ref _checkpointToken, newToken);
 
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ChildContextOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ChildContextOperation.cs
index 55c09d05f..6c97a8b20 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ChildContextOperation.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ChildContextOperation.cs
@@ -187,15 +187,30 @@ private async Task<T> ExecuteFunc(CancellationToken cancellationToken)
         T result;
         try
         {
+            // The START checkpoint may have been awaited; the workflow can have
+            // been torn down meanwhile. Never start the child function then.
+            _workflowCancellation.Token.ThrowIfCancellationRequested();
+            // Also against the clock: the awaited CONTEXT START can finish after
+            // the deadline while its timer has not run yet.
+            _workflowCancellation.ThrowIfDeadlineExceeded(
+                "The invocation deadline passed before the child context started.");
             result = await _func(childContext, linked.Token);
         }
-        catch (OperationCanceledException) when (linked.IsCancellationRequested)
+        catch (OperationCanceledException) when (
+            linked.IsCancellationRequested || _workflowCancellation.Token.IsCancellationRequested)
         {
             // Cancellation owned by the linked source — caller cancel or workflow
             // shutdown. Do NOT checkpoint CONTEXT FAIL: the termination signal
             // (or upstream cancel) owns the outcome.
             throw;
         }
+        catch (Exception) when (_workflowCancellation.IsDeadlineExceeded)
+        {
+            // The invocation deadline passed (Lambda Managed Instances only). An
+            // exception raised because work was cancelled is not a real failure:
+            // do NOT checkpoint CONTEXT FAIL. The invocation is retried.
+            throw;
+        }
         catch (NonDeterministicExecutionException)
         {
             // Replay-mismatch from an inner operation means the entire execution
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs
index 01f8c9547..0e023ebe0 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs
@@ -340,6 +340,16 @@ void SignalShortCircuit()
         // Surfacing it here means a torn-down operation propagates an
         // OperationCanceledException instead of synthesizing a spurious
         // FailureToleranceExceeded verdict from units that merely unwound.
+        //
+        // Check the workflow token itself as well (as IncrementalParallelOperation
+        // does): controlToken is linked to it through a callback, so while
+        // cancellation is still being delivered it can lag behind. A unit that
+        // failed because of the cancellation must never reach the parent checkpoint.
+        // The deadline is checked against the clock too, in case the deadline
+        // timer has not run yet when units failed because time ran out.
+        _workflowCancellation.ThrowIfDeadlineExceeded(
+            "The invocation deadline passed before the operation completed; its outcome is not recorded.");
+        _workflowCancellation.Token.ThrowIfCancellationRequested();
         controlToken.ThrowIfCancellationRequested();
 
         // Build BatchItems ONLY for units that were dispatched — matching the JS
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/DurableInvocationOptions.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/DurableInvocationOptions.cs
new file mode 100644
index 000000000..5f14140eb
--- /dev/null
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/DurableInvocationOptions.cs
@@ -0,0 +1,34 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+namespace Amazon.Lambda.DurableExecution.Internal;
+
+/// <summary>
+/// Per-invocation settings for <see cref="DurableFunction"/> that tests override.
+/// Production code uses the defaults.
+/// </summary>
+internal sealed class DurableInvocationOptions
+{
+    /// <summary>
+    /// How long checkpoint cleanup may run after the invocation deadline before
+    /// the SDK gives up on it. Only applies to Lambda Managed Instances.
+    /// </summary>
+    public static readonly TimeSpan DefaultCleanupTimeout = TimeSpan.FromSeconds(2);
+
+    /// <summary>
+    /// Overrides detection of Lambda Managed Instances. <c>null</c> reads the
+    /// <c>AWS_LAMBDA_MAX_CONCURRENCY</c> environment variable.
+    /// </summary>
+    public bool? IsMultiConcurrency { get; init; }
+
+    /// <summary>See <see cref="DefaultCleanupTimeout"/>.</summary>
+    public TimeSpan CleanupTimeout { get; init; } = DefaultCleanupTimeout;
+
+    /// <summary>
+    /// Test seam: receives the invocation's <see cref="WorkflowCancellation"/>
+    /// as soon as it is created, so a test can fire the deadline itself
+    /// (<see cref="WorkflowCancellation.TriggerDeadline"/>) at a chosen point
+    /// instead of racing a real timer.
+    /// </summary>
+    public Action<WorkflowCancellation>? OnWorkflowCancellationCreated { get; init; }
+}
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs
index 295d23e8f..72cbe74a8 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs
@@ -556,6 +556,10 @@ private async Task<IBatchResult> CompleteCoreAsync(CancellationToken cancellatio
         }
 
         // A torn-down operation propagates cancellation rather than a synthesized verdict.
+        // The deadline is checked against the clock too, in case the deadline timer
+        // has not run yet when branches failed because time ran out.
+        _workflowCancellation.ThrowIfDeadlineExceeded(
+            "The invocation deadline passed before the parallel operation completed; its outcome is not recorded.");
         _workflowCancellation.Token.ThrowIfCancellationRequested();
         cancellationToken.ThrowIfCancellationRequested();
 
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs
index 462b146bf..2dee83f69 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs
@@ -233,10 +233,18 @@ private async Task<T> ExecuteFunc(int attemptNumber, CancellationToken cancellat
                 ["attempt"] = attemptNumber,
             }))
             {
+                // The START checkpoint above may have been awaited; the workflow
+                // can have been torn down meanwhile. Never start the body then.
+                _workflowCancellation.Token.ThrowIfCancellationRequested();
+                // Also against the clock: the awaited START can finish after the
+                // deadline while its timer has not run yet.
+                _workflowCancellation.ThrowIfDeadlineExceeded(
+                    "The invocation deadline passed before the step body started.");
                 result = await _func(stepContext, linked.Token);
             }
         }
-        catch (OperationCanceledException) when (linked.IsCancellationRequested)
+        catch (OperationCanceledException) when (
+            linked.IsCancellationRequested || _workflowCancellation.Token.IsCancellationRequested)
         {
             // Cancellation owned by the linked source (caller-cancel or workflow
             // shutdown). Do NOT checkpoint FAIL and do NOT consult the retry
@@ -245,6 +253,15 @@ private async Task<T> ExecuteFunc(int attemptNumber, CancellationToken cancellat
             // as a fault on the workflow user task.
             throw;
         }
+        catch (Exception) when (_workflowCancellation.IsDeadlineExceeded)
+        {
+            // The invocation deadline passed (Lambda Managed Instances only). A
+            // body that converts its cancellation into another exception did not
+            // really fail: do NOT checkpoint FAIL or RETRY, or replay would see a
+            // permanent failure caused by a timeout. The invocation is retried
+            // and the step runs again.
+            throw;
+        }
         catch (Exception ex)
         {
             // The user step body itself threw. Funnel into the retry/fail decision
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WaitForConditionOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WaitForConditionOperation.cs
index 2345eda9d..f6768f082 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WaitForConditionOperation.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WaitForConditionOperation.cs
@@ -148,6 +148,22 @@ private Task<TState> ReplayReady(Operation ready, CancellationToken cancellation
         return ExecuteIteration(priorState, attemptNumber, existingStatus: ready.Status, cancellationToken);
     }
 
+    private void ThrowIfDeadlineExceeded() =>
+        _workflowCancellation.ThrowIfDeadlineExceeded(
+            "The invocation deadline passed during the condition check; its outcome is not recorded.");
+
+    /// <summary>
+    /// Serializes <paramref name="state"/> for a checkpoint, then re-checks the
+    /// deadline: serialization runs user serializer code, which can take long
+    /// enough to cross it. Used on every path that records an outcome.
+    /// </summary>
+    private string SerializeStateForCheckpoint(TState state)
+    {
+        var serialized = SerializeState(state);
+        ThrowIfDeadlineExceeded();
+        return serialized;
+    }
+
     private async Task<TState> ExecuteIteration(TState currentState, int attemptNumber, string? existingStatus, CancellationToken cancellationToken)
     {
         cancellationToken.ThrowIfCancellationRequested();
@@ -191,15 +207,28 @@ private async Task<TState> ExecuteIteration(TState currentState, int attemptNumb
         try
         {
             var checkContext = new ConditionCheckContext(attemptNumber, _logger);
+            // Never start the check once the workflow has been torn down.
+            _workflowCancellation.Token.ThrowIfCancellationRequested();
+            // Also against the clock: restoring the prior state on replay can
+            // cross the deadline while its timer has not run yet.
+            ThrowIfDeadlineExceeded();
             newState = await _check(currentState, checkContext, linked.Token);
         }
-        catch (OperationCanceledException) when (linked.IsCancellationRequested)
+        catch (OperationCanceledException) when (
+            linked.IsCancellationRequested || _workflowCancellation.Token.IsCancellationRequested)
         {
             // Cancellation owned by the linked source — caller cancel or workflow
             // shutdown. Do NOT checkpoint FAIL: the termination signal (or
             // upstream cancel) owns the outcome.
             throw;
         }
+        catch (Exception) when (_workflowCancellation.IsDeadlineExceeded)
+        {
+            // The invocation deadline passed (Lambda Managed Instances only). An
+            // exception raised because the check was cancelled is not a real
+            // failure: do NOT checkpoint FAIL. The invocation is retried.
+            throw;
+        }
         catch (Exception ex)
         {
             // The check threw. WaitForCondition has no per-exception retry
@@ -223,11 +252,23 @@ await EnqueueAsync(new SdkOperationUpdate
             };
         }
 
+        // The invocation deadline passed (Lambda Managed Instances only) while the
+        // check ran. A check that swallowed the cancellation and returned a state
+        // did not complete a real poll: record no RETRY (which would consume an
+        // attempt) and no FAIL. The poll runs again when the invocation is retried.
+        ThrowIfDeadlineExceeded();
+
         WaitDecision decision;
         try
         {
             decision = _config.WaitStrategy.Decide(newState, attemptNumber);
         }
+        catch (WaitForConditionException) when (_workflowCancellation.IsDeadlineExceeded)
+        {
+            // The deadline passed while the strategy ran: never record FAIL for it.
+            ThrowIfDeadlineExceeded();
+            throw;
+        }
         catch (WaitForConditionException maxEx)
         {
             // Strategy is signaling max-attempts reached. The strategy
@@ -245,6 +286,7 @@ await EnqueueAsync(new SdkOperationUpdate
             // that hits this cached FAIL can reconstruct LastState identically
             // to the live throw. The wire protocol forbids a Payload on FAIL
             // updates, so ErrorData is the only field that survives replay.
+            var lastStateData = SerializeStateForCheckpoint(newState);
             await EnqueueAsync(new SdkOperationUpdate
             {
                 Id = OperationId,
@@ -257,17 +299,21 @@ await EnqueueAsync(new SdkOperationUpdate
                 {
                     ErrorType = typeof(WaitForConditionException).FullName,
                     ErrorMessage = enriched.Message,
-                    ErrorData = SerializeState(newState)
+                    ErrorData = lastStateData
                 }
             }, cancellationToken);
 
             throw enriched;
         }
 
+        // Decide is user code too; the deadline can pass while it runs.
+        ThrowIfDeadlineExceeded();
+
         if (!decision.ShouldContinue)
         {
             // Stop() means the condition has been met. Persist the final
             // state and return it to the caller.
+            var finalState = SerializeStateForCheckpoint(newState);
             await EnqueueAsync(new SdkOperationUpdate
             {
                 Id = OperationId,
@@ -276,7 +322,7 @@ await EnqueueAsync(new SdkOperationUpdate
                 Action = OperationAction.SUCCEED,
                 SubType = OperationSubTypes.WaitForCondition,
                 Name = Name,
-                Payload = SerializeState(newState)
+                Payload = finalState
             }, cancellationToken);
 
             return newState;
@@ -287,6 +333,7 @@ await EnqueueAsync(new SdkOperationUpdate
         // service definitely has the new state and timer scheduled before
         // we suspend.
         var delaySeconds = (int)Math.Max(1, Math.Ceiling(decision.Delay.TotalSeconds));
+        var nextState = SerializeStateForCheckpoint(newState);
         await EnqueueAsync(new SdkOperationUpdate
         {
             Id = OperationId,
@@ -295,7 +342,7 @@ await EnqueueAsync(new SdkOperationUpdate
             Action = OperationAction.RETRY,
             SubType = OperationSubTypes.WaitForCondition,
             Name = Name,
-            Payload = SerializeState(newState),
+            Payload = nextState,
             StepOptions = new SdkStepOptions { NextAttemptDelaySeconds = delaySeconds }
         }, cancellationToken);
 
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WorkflowCancellation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WorkflowCancellation.cs
index 115dd585e..1d6e61d65 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WorkflowCancellation.cs
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/WorkflowCancellation.cs
@@ -1,14 +1,36 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+using System.Diagnostics;
+using Amazon.Lambda.Core;
+
 namespace Amazon.Lambda.DurableExecution.Internal;
 
+/// <summary>
+/// Why the workflow-scoped token was cancelled. The first reason wins.
+/// </summary>
+internal enum WorkflowCancellationReason
+{
+    None = 0,
+
+    /// <summary>The <see cref="TerminationManager"/> resolved (suspend or checkpoint failure).</summary>
+    Terminated = 1,
+
+    /// <summary>The invocation deadline passed (Lambda Managed Instances only).</summary>
+    DeadlineExceeded = 2,
+
+    /// <summary>The wrapper finished the invocation and is tearing it down.</summary>
+    Teardown = 3,
+}
+
 /// <summary>
 /// Workflow-scoped cancellation source. Cancels when the
-/// <see cref="TerminationManager"/> resolves so abandoned user-<c>Func</c> bodies
-/// (the WhenAny loser in <see cref="DurableExecutionHandler"/>) unwind via
+/// <see cref="TerminationManager"/> resolves, when the optional invocation
+/// deadline passes, or when <see cref="DurableFunction"/> tears the invocation
+/// down. Abandoned user-<c>Func</c> bodies (the WhenAny loser in
+/// <see cref="DurableExecutionHandler"/>) unwind via
 /// <see cref="OperationCanceledException"/> instead of running to completion on
-/// the threadpool while Lambda is mid-response.
+/// the threadpool.
 /// </summary>
 /// <remarks>
 /// One instance per durable function invocation, constructed and disposed by
@@ -20,27 +42,371 @@ namespace Amazon.Lambda.DurableExecution.Internal;
 /// observe this token: successful work must persist even when the workflow is
 /// being torn down.
 /// </para>
+/// <para>
+/// Rules that matter when several invocations share one process (Lambda
+/// Managed Instances), where a stuck SDK thread blocks a runtime processing loop:
+/// </para>
+/// <list type="bullet">
+///   <item>The SDK never runs cancellation callbacks it does not own on its own
+///       threads. User code and the service client register callbacks on these
+///       tokens (directly, or through linked tokens), and a callback can block or
+///       throw. Every cancel the SDK starts uses
+///       <see cref="CancellationTokenSource.CancelAsync"/>: the token reports
+///       <c>IsCancellationRequested</c> immediately, and callbacks run on the
+///       threadpool. Their exceptions are observed and logged.</item>
+///   <item>The SDK's own decisions are published before any foreign code runs:
+///       admission is closed, the deadline is recorded and
+///       <see cref="DeadlineTask"/> is signalled first.</item>
+///   <item>The deadline is a monotonic timestamp, not just a timer. A timer
+///       callback can run late when the threadpool is busy, so
+///       <see cref="IsDeadlineExceeded"/> also compares the clock, and fires the
+///       deadline itself when the timer has not yet.</item>
+///   <item>Cancellation sources are disposed only after callback delivery has
+///       finished, so no registered callback is ever skipped. <see cref="Token"/>
+///       is captured at construction, so abandoned user code that reads it after
+///       disposal gets a cancelled token instead of an
+///       <see cref="ObjectDisposedException"/>.</item>
+/// </list>
 /// </remarks>
 internal sealed class WorkflowCancellation : IDisposable
 {
     private readonly CancellationTokenSource _cts = new();
+    private readonly CancellationToken _token;
+    private readonly CancellationTokenSource? _deadlineCts;
+    private readonly long _deadlineTimestamp;
+    private readonly Timer? _deadlineTimer;
+    private readonly TaskCompletionSource _deadlineTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+    private readonly object _cancelLock = new();
+    private bool _sourcesReleased;
+    private Task? _cancelDelivery;
+    private Task? _deadlineDelivery;
+    private int _reason;
+    private int _tornDown;
+    private int _deadlineFired;
+    private int _disposed;
+
+    public CancellationToken Token => _token;
+
+    /// <summary>
+    /// The reason the token was cancelled, or <see cref="WorkflowCancellationReason.None"/>.
+    /// </summary>
+    public WorkflowCancellationReason Reason => (WorkflowCancellationReason)Volatile.Read(ref _reason);
+
+    /// <summary>
+    /// True once the invocation deadline passed. Compares the clock as well as
+    /// checking whether the deadline fired, so it is accurate even when the
+    /// threadpool delays the timer or a suspension cancelled the workflow first.
+    /// When the clock shows the deadline has passed but the timer has not run,
+    /// this fires the deadline, so every observer sees one consistent outcome.
+    /// </summary>
+    public bool IsDeadlineExceeded
+    {
+        get
+        {
+            if (_deadlineCts is null) return false;
+            if (Volatile.Read(ref _deadlineFired) == 1) return true;
+            if (Stopwatch.GetTimestamp() < _deadlineTimestamp) return false;
+            TriggerDeadline();
+            return true;
+        }
+    }
+
+    /// <summary>
+    /// True once the invocation is being torn down (the wrapper finished or the
+    /// deadline passed). New durable operations are rejected from this point.
+    /// </summary>
+    public bool IsTornDown => Volatile.Read(ref _tornDown) == 1 || IsDeadlineExceeded;
+
+    /// <summary>
+    /// Cancels when the invocation deadline passes. Passed to SDK-owned service
+    /// calls (state hydration). <see cref="CancellationToken.None"/> when there is
+    /// no deadline, so on-demand Lambda calls are unchanged. Do not wait on this
+    /// token to learn the deadline passed: its callbacks belong to the client and
+    /// may be delayed. Use <see cref="DeadlineTask"/> or <see cref="IsDeadlineExceeded"/>.
+    /// </summary>
+    public CancellationToken DeadlineToken { get; }
+
+    /// <summary>
+    /// Completes when the invocation deadline passes, before any foreign callback
+    /// runs. Never completes when there is no deadline.
+    /// </summary>
+    public Task DeadlineTask => _deadlineTcs.Task;
 
-    public CancellationToken Token => _cts.Token;
+    /// <summary>
+    /// Completes once every callback registered on <see cref="Token"/> has run.
+    /// Already complete before the token is cancelled. Exposed for tests and
+    /// for disposal ordering.
+    /// </summary>
+    internal Task CancellationDelivered => Volatile.Read(ref _cancelDelivery) ?? Task.CompletedTask;
 
-    public WorkflowCancellation(TerminationManager terminationManager)
+    /// <param name="terminationManager">Suspension signal for this invocation.</param>
+    /// <param name="deadline">
+    /// Time until the invocation deadline, or <c>null</c> for no deadline. Only
+    /// set for Lambda Managed Instances; see <see cref="InvocationDeadline"/>.
+    /// </param>
+    /// <param name="anchorTimestamp">
+    /// The <see cref="Stopwatch.GetTimestamp"/> at which <paramref name="deadline"/>
+    /// was measured. Time elapsed since then is subtracted, so the deadline does
+    /// not drift by however long it took to get here. Defaults to now.
+    /// </param>
+    /// <param name="armTimer">
+    /// Test seam. <c>false</c> leaves the deadline to fire only through
+    /// <see cref="TriggerDeadline"/> or the clock check in
+    /// <see cref="IsDeadlineExceeded"/>, which models a timer the threadpool has
+    /// not run yet. Production code always arms the timer.
+    /// </param>
+    public WorkflowCancellation(TerminationManager terminationManager, TimeSpan? deadline = null, long? anchorTimestamp = null,
+        bool armTimer = true)
     {
+        _token = _cts.Token;
+
         terminationManager.TerminationTask.ContinueWith(
-            static (_, state) =>
+            static (_, state) => ((WorkflowCancellation)state!).Cancel(WorkflowCancellationReason.Terminated),
+            this,
+            CancellationToken.None,
+            TaskContinuationOptions.ExecuteSynchronously,
+            TaskScheduler.Default);
+
+        if (deadline is { } remaining)
+        {
+            _deadlineCts = new CancellationTokenSource();
+            DeadlineToken = _deadlineCts.Token;
+
+            var anchor = anchorTimestamp ?? Stopwatch.GetTimestamp();
+            _deadlineTimestamp = anchor + (long)(remaining.TotalSeconds * Stopwatch.Frequency);
+            var untilDeadline = remaining - Stopwatch.GetElapsedTime(anchor);
+
+            // A deadline already reached (including one that ran out between
+            // reading RemainingTime and getting here) fires at once.
+            if (!armTimer)
             {
-                var cts = (CancellationTokenSource)state!;
-                try { cts.Cancel(); }
-                catch (ObjectDisposedException) { }
+                // Test seam: nothing fires the deadline on its own.
+            }
+            else if (untilDeadline > TimeSpan.Zero)
+                _deadlineTimer = new Timer(static state => ((WorkflowCancellation)state!).TriggerDeadline(), this, untilDeadline, Timeout.InfiniteTimeSpan);
+            else
+                TriggerDeadline();
+        }
+    }
+
+    /// <summary>
+    /// Marks the invocation as torn down and cancels the token. Returns without
+    /// running user cancellation callbacks: they run on the threadpool, so one
+    /// that blocks cannot stall the wrapper before its bounded drain.
+    /// </summary>
+    public void BeginTeardown()
+    {
+        Volatile.Write(ref _tornDown, 1);
+        Cancel(WorkflowCancellationReason.Teardown);
+    }
+
+    /// <summary>
+    /// Throws <see cref="OperationCanceledException"/> when the invocation is
+    /// torn down, so abandoned user code cannot start new durable operations.
+    /// </summary>
+    public void ThrowIfTornDown()
+    {
+        if (IsTornDown)
+            throw new OperationCanceledException(
+                "The durable invocation has ended; no new durable operations can be started.", _token);
+    }
+
+    /// <summary>
+    /// Throws <see cref="OperationCanceledException"/> when the invocation
+    /// deadline has passed, so a timed-out invocation records no outcome.
+    /// </summary>
+    public void ThrowIfDeadlineExceeded(string message)
+    {
+        if (IsDeadlineExceeded)
+            throw new OperationCanceledException(message, _token);
+    }
+
+    /// <summary>
+    /// Fires the deadline now: by the timer, by an observer that saw the clock
+    /// pass it, or by a test. Idempotent. Runs no foreign code on the caller's
+    /// thread, so it is safe to call from anywhere.
+    /// </summary>
+    internal void TriggerDeadline()
+    {
+        if (_deadlineCts is null) return;
+        if (Interlocked.Exchange(ref _deadlineFired, 1) != 0) return;
+
+        // The SDK's own state first: record why and close admission.
+        Interlocked.CompareExchange(ref _reason, (int)WorkflowCancellationReason.DeadlineExceeded, 0);
+        Volatile.Write(ref _tornDown, 1);
+
+        // Then start cancelling the workflow token and the client's deadline
+        // token. Both only queue their callbacks, so no foreign code runs here.
+        // This must happen BEFORE DeadlineTask is signalled: a waiter on
+        // DeadlineTask (the hydration wait) can end the invocation and reach
+        // Dispose at once, and Dispose must find the deadline cancel already
+        // under way, not dispose a source whose cancel has not started.
+        Cancel(WorkflowCancellationReason.DeadlineExceeded);
+        StartCancelOnce(ref _deadlineDelivery, _deadlineCts,
+            "A cancellation callback registered on the durable invocation deadline threw.");
+
+        // Last, signal the handler. _deadlineTcs runs its continuations
+        // asynchronously, so this never runs the handler on the calling thread.
+        _deadlineTcs.TrySetResult();
+        AfterDeadlineSignalledForTest?.Invoke();
+    }
+
+    /// <summary>
+    /// Test seam: runs right after <see cref="DeadlineTask"/> is signalled, on the
+    /// thread firing the deadline. Lets a test act (for example dispose) at the
+    /// earliest point a waiter on <see cref="DeadlineTask"/> could.
+    /// </summary>
+    internal Action? AfterDeadlineSignalledForTest { get; set; }
+
+    private void Cancel(WorkflowCancellationReason reason)
+    {
+        Interlocked.CompareExchange(ref _reason, (int)reason, 0);
+        StartCancelOnce(ref _cancelDelivery, _cts,
+            "A cancellation callback registered by the durable workflow threw while the workflow was being cancelled.");
+    }
+
+    /// <summary>
+    /// Starts cancelling <paramref name="source"/> exactly once, and records the
+    /// delivery so <see cref="Dispose"/> can wait for it.
+    /// </summary>
+    /// <remarks>
+    /// The lock makes every caller return only after the source reports
+    /// <c>IsCancellationRequested</c>: a caller that loses the race waits for the
+    /// winner to start cancelling instead of returning early. It is held only
+    /// while <see cref="CancellationTokenSource.CancelAsync"/> flips the state and
+    /// queues the callbacks, so no foreign code ever runs under it.
+    /// </remarks>
+    private void StartCancelOnce(ref Task? deliverySlot, CancellationTokenSource source, string callbackFailureMessage)
+    {
+        lock (_cancelLock)
+        {
+            // Once Dispose has taken the delivery handles, the sources may be
+            // disposed; a cancel that had not started by then never starts.
+            if (deliverySlot != null || _sourcesReleased) return;
+            Volatile.Write(ref deliverySlot, StartCancel(source, callbackFailureMessage));
+        }
+    }
+
+    /// <summary>
+    /// Cancels <paramref name="source"/> without running its callbacks on this
+    /// thread. The returned task completes when every callback has run; callback
+    /// exceptions are logged, never rethrown.
+    /// </summary>
+    private static Task StartCancel(CancellationTokenSource source, string callbackFailureMessage)
+    {
+        Task delivery;
+        try
+        {
+            delivery = source.CancelAsync();
+        }
+        catch (ObjectDisposedException)
+        {
+            // Only reachable after disposal, which happens once delivery finished,
+            // so the source was already cancelled.
+            return Task.CompletedTask;
+        }
+
+        return delivery.ContinueWith(
+            static (t, message) =>
+            {
+                if (t.Exception is { } ex)
+                    LambdaLogger.Log("Warning", ex.Flatten(), (string)message!);
             },
-            _cts,
+            callbackFailureMessage,
             CancellationToken.None,
             TaskContinuationOptions.ExecuteSynchronously,
             TaskScheduler.Default);
     }
 
-    public void Dispose() => _cts.Dispose();
+    public void Dispose()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+
+        BeginTeardown();
+        _deadlineTimer?.Dispose();
+
+        // Dispose each source only after its callbacks have all run. Disposing
+        // earlier could skip callbacks that have not run yet, so linked tokens
+        // (and the user code behind them) would never be cancelled. A callback
+        // that never returns just leaves its source undisposed. Taking the
+        // delivery handles and closing out new cancels happen together under the
+        // lock: a cancel is either already under way (and is waited for) or never
+        // starts, so a source is never disposed while its cancel is starting.
+        Task workflowDelivery;
+        Task? deadlineDelivery;
+        lock (_cancelLock)
+        {
+            _sourcesReleased = true;
+            workflowDelivery = _cancelDelivery ?? Task.CompletedTask;
+            deadlineDelivery = _deadlineDelivery;
+        }
+
+        DisposeAfter(_cts, workflowDelivery);
+        if (_deadlineCts != null)
+        {
+            // A deadline that never fired was never cancelled: dispose it now. A
+            // timer callback still in flight finds _sourcesReleased set and does
+            // not touch the source.
+            DisposeAfter(_deadlineCts, deadlineDelivery ?? Task.CompletedTask);
+        }
+    }
+
+    private static void DisposeAfter(CancellationTokenSource source, Task delivery)
+    {
+        if (delivery.IsCompleted)
+            source.Dispose();
+        else
+            delivery.ContinueWith(
+                static (_, s) => ((CancellationTokenSource)s!).Dispose(),
+                source,
+                CancellationToken.None,
+                TaskContinuationOptions.ExecuteSynchronously,
+                TaskScheduler.Default);
+    }
+}
+
+/// <summary>
+/// Decides whether an invocation gets a deadline that cancels durable work.
+/// </summary>
+/// <remarks>
+/// Only Lambda Managed Instances get a deadline. There, several invocations
+/// share one process and Lambda does not stop code when an invocation times out,
+/// so the SDK has to stop its own work. In on-demand Lambda the execution
+/// environment is stopped at the timeout, so the SDK keeps its existing
+/// behavior and never cancels on a deadline.
+/// </remarks>
+internal static class InvocationDeadline
+{
+    /// <summary>
+    /// Environment variable Lambda sets for functions running on Lambda Managed
+    /// Instances (multi-concurrency mode).
+    /// </summary>
+    internal const string MaxConcurrencyEnvironmentVariable = "AWS_LAMBDA_MAX_CONCURRENCY";
+
+    /// <summary>True when the function runs on Lambda Managed Instances.</summary>
+    public static bool IsMultiConcurrency() =>
+        !string.IsNullOrEmpty(Environment.GetEnvironmentVariable(MaxConcurrencyEnvironmentVariable));
+
+    /// <summary>
+    /// Returns the time until the deadline, <see cref="TimeSpan.Zero"/> when the
+    /// deadline has already passed, or <c>null</c> for no deadline.
+    /// </summary>
+    /// <param name="remainingTime">The invocation's <c>ILambdaContext.RemainingTime</c>.</param>
+    /// <param name="isMultiConcurrency">Whether the function runs on Lambda Managed Instances.</param>
+    /// <remarks>
+    /// On Lambda Managed Instances a zero or negative remaining time means the
+    /// invocation was already out of time when it started, so it times out
+    /// immediately. Tests that exercise this mode must set
+    /// <c>TestLambdaContext.RemainingTime</c>.
+    /// </remarks>
+    public static TimeSpan? Resolve(TimeSpan remainingTime, bool isMultiConcurrency)
+    {
+        if (!isMultiConcurrency) return null;
+        if (remainingTime <= TimeSpan.Zero) return TimeSpan.Zero;
+
+        // Timer rejects delays above int.MaxValue milliseconds. Lambda timeouts
+        // are far below that; treat anything larger as no deadline.
+        if (remainingTime.TotalMilliseconds >= int.MaxValue) return null;
+        return remainingTime;
+    }
 }
diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/cancellation.md b/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/cancellation.md
index 1abc75858..ef3da6beb 100644
--- a/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/cancellation.md
+++ b/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/cancellation.md
@@ -22,6 +22,22 @@ When either trigger fires, the token transitions to `IsCancellationRequested = t
 - **`OperationCanceledException` thrown out of a step body via the token** (i.e. `linked.IsCancellationRequested` is true) is treated as cancellation: no FAIL checkpoint is written, no retry is consulted. The exception propagates up.
 - **`OperationCanceledException` thrown by user code for unrelated reasons** (token never fired) is treated as a normal step failure: FAIL checkpoint, retry per the configured `RetryStrategy`.
 - **The SDK's own writes (checkpoints, batcher flush, the runtime API response)** never observe the workflow-shutdown signal. Successful work is never lost to teardown.
+- **Callbacks you register on the token** (`ct.Register(...)`) run on the thread pool, not on the thread that cancelled. The token itself reports `IsCancellationRequested` immediately. A callback that throws is logged, not rethrown.
+- **When the invocation ends**, the token is cancelled before `DurableFunction.WrapAsync` returns. Work the workflow started but did not await (for example a step still running when a sibling suspended) observes the cancellation. Starting a new durable operation after this point throws `OperationCanceledException`.
+
+## Lambda Managed Instances
+
+On Lambda Managed Instances several invocations run in the same process, and Lambda does not stop an invocation that times out. The SDK therefore applies the invocation deadline (`ILambdaContext.RemainingTime`) itself:
+
+- At the deadline the token is cancelled, new durable operations are rejected, and state loading is cancelled.
+- An exception your code throws after the deadline (for example, after catching the cancellation) is not recorded as a step failure, and does not consume a retry.
+- An invocation that starts with its deadline already passed times out immediately. When testing in this mode, set `TestLambdaContext.RemainingTime`.
+- Checkpoint cleanup gets a short budget past the deadline, then is abandoned.
+- `WrapAsync` throws `TimeoutException`. Lambda treats this as an invocation error, the same outcome as a timeout in on-demand Lambda: the execution is not marked failed, and work that was not checkpointed runs again on replay.
+
+Only cooperative code can be stopped. Code that ignores the token, or blocks synchronously, keeps running until it returns, and keeps a thread busy that other invocations could use.
+
+On-demand Lambda is unchanged: the SDK applies no deadline, and the environment's timeout ends the invocation.
 
 ## Guidance
 
diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/LambdaManagedInstancesTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/LambdaManagedInstancesTests.cs
new file mode 100644
index 000000000..d430e85c9
--- /dev/null
+++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/LambdaManagedInstancesTests.cs
@@ -0,0 +1,1680 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+using System.Collections.Concurrent;
+using Amazon.Lambda.DurableExecution.Internal;
+using Amazon.Lambda.DurableExecution.Services;
+using Amazon.Lambda.Serialization.SystemTextJson;
+using Amazon.Lambda.TestUtilities;
+using Xunit;
+using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate;
+
+namespace Amazon.Lambda.DurableExecution.Tests;
+
+/// <summary>
+/// Tests for running durable functions on Lambda Managed Instances (LMI), where
+/// several invocations run in one process at the same time and Lambda does not
+/// stop an invocation that times out.
+/// </summary>
+/// <remarks>
+/// These tests drive <see cref="DurableFunction"/> directly with an in-memory
+/// service client, running invocations concurrently the way RuntimeSupport does
+/// in multi-concurrency mode. LMI mode is selected through
+/// <see cref="DurableInvocationOptions.IsMultiConcurrency"/> rather than the
+/// process-wide environment variable, so these tests can run in parallel.
+/// Ordering is controlled with signals, never with sleeps; the short deadlines
+/// used below only need to expire eventually, and no test asserts on elapsed time.
+/// </remarks>
+public class LambdaManagedInstancesTests
+{
+    // Guards against a hang turning into a stuck test run. Not a timing assertion.
+    private static readonly TimeSpan HangGuard = TimeSpan.FromSeconds(30);
+
+    private static readonly DurableInvocationOptions LmiOptions = new() { IsMultiConcurrency = true };
+    private static readonly DurableInvocationOptions OnDemandOptions = new() { IsMultiConcurrency = false };
+
+    private static string IdAt(int position) => OperationIdGenerator.HashOperationId(position.ToString());
+
+    // ── Teardown ordering (all modes) ──────────────────────────────────
+
+    [Fact]
+    public void WorkflowCancellation_TerminateThenDispose_TokenIsAlwaysCancelled()
+    {
+        // Termination delivers its cancel through a threadpool continuation.
+        // Disposing right after Terminate used to win that race and drop the
+        // cancel. Repeat to make the old race show up reliably.
+        for (var i = 0; i < 500; i++)
+        {
+            var tm = new TerminationManager();
+            var wfc = new WorkflowCancellation(tm);
+            var token = wfc.Token;
+
+            tm.Terminate(TerminationReason.WaitScheduled);
+            wfc.Dispose();
+
+            Assert.True(token.IsCancellationRequested, $"Cancellation lost on iteration {i}.");
+        }
+    }
+
+    [Fact]
+    public void WorkflowCancellation_TokenIsUsableAfterDispose()
+    {
+        var wfc = new WorkflowCancellation(new TerminationManager());
+        wfc.Dispose();
+
+        // Abandoned user code reads the token after the invocation ends. It must
+        // get a cancelled token, not an ObjectDisposedException.
+        var token = wfc.Token;
+        Assert.True(token.IsCancellationRequested);
+        using var linked = CancellationTokenSource.CreateLinkedTokenSource(CancellationToken.None, token);
+        Assert.True(linked.IsCancellationRequested);
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Suspended_AbandonedStepTokenIsCancelledBeforeWrapperReturns(bool isMultiConcurrency)
+    {
+        // Repeat to cover warm, back-to-back invocations of the same handler.
+        for (var i = 0; i < 20; i++)
+        {
+            var client = new FakeServiceClient();
+            var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            CancellationToken stepToken = default;
+
+            var output = await DurableFunction.WrapAsync<Payload, string>(
+                async (input, ctx) =>
+                {
+                    // Not awaited: this step is still running when the workflow suspends.
+                    ctx.StepAsync<int>(async (_, ct) =>
+                    {
+                        stepToken = ct;
+                        entered.TrySetResult();
+                        await Task.Delay(Timeout.Infinite, ct);
+                        return 0;
+                    }, name: "slow").Observe();
+                    await entered.Task;
+
+                    await ctx.WaitAsync(TimeSpan.FromSeconds(30), name: "wait");
+                    return "unreachable";
+                },
+                Input($"arn:abandoned-{i}"),
+                Context(TimeSpan.FromMinutes(5)),
+                client,
+                new DurableInvocationOptions { IsMultiConcurrency = isMultiConcurrency });
+
+            Assert.Equal(InvocationStatus.Pending, output.Status);
+            // Cancellation is requested before the wrapper returns, and its
+            // callbacks are delivered on the threadpool (never on the wrapper's
+            // thread, where a blocking user callback could stall it). It must
+            // arrive, never be lost to disposal.
+            var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            stepToken.Register(() => cancelled.TrySetResult());
+            await cancelled.Task.NoHang();
+        }
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task AfterTeardown_AbandonedCodeCannotStartNewOperations(bool isMultiConcurrency)
+    {
+        var client = new FakeServiceClient();
+        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var bodyRan = false;
+        Task<int>? leaked = null;
+
+        var output = await DurableFunction.WrapAsync<Payload, string>(
+            async (_, ctx) =>
+            {
+                // Work the workflow started but never awaited; it resumes after the
+                // invocation has ended.
+                leaked = Task.Run(async () =>
+                {
+                    await release.Task;
+                    return await ctx.StepAsync<int>(async (_, _) =>
+                    {
+                        bodyRan = true;
+                        await Task.CompletedTask;
+                        return 1;
+                    }, name: "late");
+                });
+
+                await ctx.WaitAsync(TimeSpan.FromSeconds(30), name: "wait");
+                return "unreachable";
+            },
+            Input("arn:leaked"),
+            Context(TimeSpan.FromMinutes(5)),
+            client,
+            new DurableInvocationOptions { IsMultiConcurrency = isMultiConcurrency });
+
+        Assert.Equal(InvocationStatus.Pending, output.Status);
+        var checkpointsAtReturn = client.Calls.Count;
+
+        release.SetResult();
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => leaked!.NoHang());
+
+        Assert.False(bodyRan);
+        Assert.Equal(checkpointsAtReturn, client.Calls.Count);
+    }
+
+    // ── Deadline: on-demand is unchanged ───────────────────────────────
+
+    [Theory]
+    [InlineData(false, 300_000, null)]   // on-demand: never a deadline
+    [InlineData(false, 1, null)]
+    [InlineData(true, 0, 0)]             // LMI and already expired: time out at once
+    [InlineData(true, -5, 0)]
+    [InlineData(true, 300_000, 300_000)]
+    public void InvocationDeadline_Resolve(bool isMultiConcurrency, long remainingMs, long? expectedMs)
+    {
+        var deadline = InvocationDeadline.Resolve(TimeSpan.FromMilliseconds(remainingMs), isMultiConcurrency);
+        Assert.Equal(expectedMs, deadline is { } d ? (long)d.TotalMilliseconds : null);
+    }
+
+    [Fact]
+    public void InvocationDeadline_Resolve_IgnoresValuesBeyondTimerRange()
+    {
+        Assert.Null(InvocationDeadline.Resolve(TimeSpan.FromDays(60), isMultiConcurrency: true));
+    }
+
+    [Fact]
+    public async Task OnDemand_HydrationIsNotCancellable()
+    {
+        var client = new FakeServiceClient();
+
+        var output = await DurableFunction.WrapAsync<Payload, string>(
+            (input, _) => Task.FromResult(input.Id!),
+            Input("arn:hydrate-od", nextMarker: "page-2"),
+            Context(TimeSpan.FromMilliseconds(1)),
+            client,
+            OnDemandOptions);
+
+        Assert.Equal(InvocationStatus.Succeeded, output.Status);
+        Assert.False(Assert.Single(client.HydrationTokens).CanBeCanceled);
+    }
+
+    [Fact]
+    public async Task Lmi_HydrationReceivesDeadlineToken()
+    {
+        var client = new FakeServiceClient();
+
+        var output = await DurableFunction.WrapAsync<Payload, string>(
+            (input, _) => Task.FromResult(input.Id!),
+            Input("arn:hydrate-lmi", nextMarker: "page-2"),
+            Context(TimeSpan.FromMinutes(5)),
+            client,
+            LmiOptions);
+
+        Assert.Equal(InvocationStatus.Succeeded, output.Status);
+        Assert.True(Assert.Single(client.HydrationTokens).CanBeCanceled);
+    }
+
+    // ── Deadline: LMI ──────────────────────────────────────────────────
+
+    [Fact]
+    public async Task Lmi_DeadlinePassesDuringStep_ThrowsTimeout_WithoutFailCheckpoint()
+    {
+        // A controllable deadline, fired only once the step body is running, so
+        // the test always exercises a deadline during the step. A real short
+        // deadline could expire before the queued workflow even starts, which the
+        // wrapper validly rejects without running the body.
+        var client = new FakeServiceClient();
+        CancellationToken stepToken = default;
+        var stepEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        WorkflowCancellation? wfc = null;
+        var options = new DurableInvocationOptions
+        {
+            IsMultiConcurrency = true,
+            OnWorkflowCancellationCreated = created => wfc = created,
+        };
+
+        var run = DurableFunction.WrapAsync<Payload, string>(
+            async (_, ctx) => await ctx.StepAsync<string>(async (_, ct) =>
+            {
+                stepToken = ct;
+                stepEntered.TrySetResult();
+                await Task.Delay(Timeout.Infinite, ct);
+                return "unreachable";
+            }, name: "slow"),
+            Input("arn:deadline-step"),
+            Context(ControlledDeadline),
+            client,
+            options);
+
+        await stepEntered.Task.NoHang();
+        wfc!.TriggerDeadline();
+        var ex = await Assert.ThrowsAsync<TimeoutException>(() => run.NoHang());
+
+        Assert.Contains("arn:deadline-step", ex.Message);
+        Assert.True(stepToken.IsCancellationRequested);
+        // A timeout must not be recorded as a durable failure; replay retries it.
+        Assert.DoesNotContain(client.AllUpdates, u => u.Action == OperationAction.FAIL);
+    }
+
+    [Fact]
+    public async Task Lmi_WorkflowConvertsCancellationToOwnError_StillReportsTimeout()
+    {
+        // User code that catches the cancellation and throws its own exception
+        // must not turn a timeout into a permanently failed execution.
+        var client = new FakeServiceClient();
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                async (_, ctx) =>
+                {
+                    try
+                    {
+                        return await ctx.StepAsync<string>(async (_, ct) =>
+                        {
+                            await Task.Delay(Timeout.Infinite, ct);
+                            return "unreachable";
+                        }, name: "slow");
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        throw new InvalidOperationException("user code gave up");
+                    }
+                },
+                Input("arn:deadline-user-catch"),
+                Context(TimeSpan.FromMilliseconds(100)),
+                client,
+                LmiOptions).NoHang());
+    }
+
+    [Fact]
+    public async Task RunAsync_WorkflowFaultsFromDeadlineBeforeDeadlineTaskCompletes_ReportsDeadline()
+    {
+        // The workflow faults while the deadline has passed but DeadlineTask is
+        // not yet signalled: the deadline is already behind the clock and its
+        // timer has not run (as when the threadpool delays it). The workflow wins
+        // the race; the fault must still be reported as a deadline, not as a
+        // durable failure.
+        var tm = new TerminationManager();
+        var (wfc, deadlinePassed) = DeadlineWithTimerNotRun(tm);
+        using var _ = wfc;
+        var workflowRan = false;
+
+        var result = await DurableExecutionHandler.RunAsync<string>(
+            new ExecutionState(),
+            tm,
+            async () =>
+            {
+                workflowRan = true;
+                await deadlinePassed();
+                throw new InvalidOperationException("user gave up");
+            },
+            wfc).NoHang();
+
+        Assert.True(workflowRan);
+        Assert.True(result.DeadlineExceeded);
+        Assert.Null(result.Exception);
+    }
+
+    // ── Copilot review round 1 ─────────────────────────────────────────
+
+    [Fact]
+    public async Task Lmi_AlreadyExpiredInvocation_TimesOutWithoutRunningWorkflow()
+    {
+        var client = new FakeServiceClient();
+        var ran = false;
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                (input, _) => { ran = true; return Task.FromResult(input.Id!); },
+                Input("arn:expired"),
+                Context(TimeSpan.FromMilliseconds(-1)),
+                client,
+                LmiOptions).NoHang());
+
+        Assert.False(ran);
+        Assert.Empty(client.Calls);
+    }
+
+    [Fact]
+    public async Task WorkflowCancellation_DeadlineIsReported_EvenWhenTerminationCancelledFirst()
+    {
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var deadlineHit = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        wfc.DeadlineToken.Register(() => deadlineHit.TrySetResult());
+
+        // A suspension sets the first cancellation reason before the deadline.
+        wfc.BeginTeardown();
+        wfc.TriggerDeadline();
+        await deadlineHit.Task.NoHang();
+
+        Assert.True(wfc.IsDeadlineExceeded);
+    }
+
+    [Fact]
+    public async Task WorkflowCancellation_ThrowingCallback_DoesNotEscapeDeadlineOrTeardown()
+    {
+        var tm = new TerminationManager();
+        var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        // Callbacks run in reverse registration order: the throwing one runs
+        // first, then this one. It still running proves every callback was
+        // delivered despite the throw.
+        var otherCallbackRan = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        wfc.Token.Register(() => otherCallbackRan.TrySetResult());
+        wfc.Token.Register(() => throw new InvalidOperationException("user callback threw"));
+
+        // Firing the deadline must not throw on the caller's thread (on the timer
+        // thread that would crash the process) or block the signal.
+        wfc.TriggerDeadline();
+        await wfc.DeadlineTask.NoHang();
+        await otherCallbackRan.Task.NoHang();
+        await wfc.CancellationDelivered.NoHang();
+        Assert.True(wfc.Token.IsCancellationRequested);
+
+        // Teardown and dispose must not throw either.
+        wfc.BeginTeardown();
+        wfc.Dispose();
+    }
+
+    [Fact]
+    public async Task WorkflowCancellation_ThrowingCallback_DoesNotEscapeTeardown()
+    {
+        // No deadline: the first cancel comes from the wrapper's teardown, as on
+        // every invocation. A throwing user callback must not replace the
+        // invocation's outcome.
+        var wfc = new WorkflowCancellation(new TerminationManager());
+        var otherCallbackRan = false;
+        wfc.Token.Register(() => otherCallbackRan = true);
+        wfc.Token.Register(() => throw new InvalidOperationException("user callback threw"));
+
+        wfc.BeginTeardown();
+        await wfc.CancellationDelivered.NoHang();
+        wfc.Dispose();
+
+        Assert.True(wfc.Token.IsCancellationRequested);
+        Assert.True(otherCallbackRan);
+    }
+
+    [Fact]
+    public async Task WorkflowCancellation_BlockingCallback_DoesNotDelayDeadlineSignal()
+    {
+        var tm = new TerminationManager();
+        using var release = new ManualResetEventSlim();
+        var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        wfc.Token.Register(() => release.Wait());
+
+        try
+        {
+            // Firing returns at once and DeadlineTask completes, even though a
+            // cancellation callback blocks the thread delivering the cancel.
+            await Task.Run(wfc.TriggerDeadline).NoHang();
+            await wfc.DeadlineTask.NoHang();
+            Assert.True(wfc.IsDeadlineExceeded);
+        }
+        finally
+        {
+            release.Set();
+            wfc.Dispose();
+        }
+    }
+
+    // ── Copilot review round 3 ─────────────────────────────────────────
+
+    [Fact]
+    public async Task WorkflowCancellation_BeginTeardown_DoesNotRunUserCallbacksOnCaller()
+    {
+        // A step's linked token carries a user callback that blocks. Teardown is
+        // the first cancel; it must return without waiting for that callback.
+        var wfc = new WorkflowCancellation(new TerminationManager());
+        using var release = new ManualResetEventSlim();
+        using var stepToken = CancellationTokenSource.CreateLinkedTokenSource(CancellationToken.None, wfc.Token);
+        stepToken.Token.Register(() => release.Wait());
+
+        try
+        {
+            await Task.Run(wfc.BeginTeardown).NoHang();
+            // The token reports cancellation immediately; delivery is still pending.
+            Assert.True(wfc.Token.IsCancellationRequested);
+            Assert.False(wfc.CancellationDelivered.IsCompleted);
+        }
+        finally
+        {
+            release.Set();
+        }
+
+        await wfc.CancellationDelivered.NoHang();
+        Assert.True(stepToken.IsCancellationRequested);
+        wfc.Dispose();
+    }
+
+    [Fact]
+    public async Task OnDemand_UnawaitedStepWithBlockingCallback_WrapperStillReturns()
+    {
+        // End to end, in on-demand mode: the workflow returns while an unawaited
+        // step has a blocking callback on its token. Teardown runs on every
+        // invocation and must not stall in that callback.
+        var client = new FakeServiceClient();
+        using var release = new ManualResetEventSlim();
+        var registered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        try
+        {
+            var output = await DurableFunction.WrapAsync<Payload, string>(
+                async (_, ctx) =>
+                {
+                    ctx.StepAsync<int>(async (_, ct) =>
+                    {
+                        ct.Register(() => release.Wait());
+                        registered.TrySetResult();
+                        await Task.Delay(Timeout.Infinite, ct);
+                        return 0;
+                    }, name: "slow").Observe();
+                    await registered.Task;
+                    return "done";
+                },
+                Input("arn:blocking-teardown"),
+                Context(TimeSpan.FromMinutes(5)),
+                client,
+                OnDemandOptions).NoHang();
+
+            Assert.Equal(InvocationStatus.Succeeded, output.Status);
+        }
+        finally
+        {
+            release.Set();
+        }
+    }
+
+    [Fact]
+    public void WorkflowCancellation_DeadlinePassedButTimerNotRun_IsReportedAndClosesAdmission()
+    {
+        var tm = new TerminationManager();
+        using var wfc = ExpiredWithTimerNotRun(tm);
+
+        // Nothing has fired the deadline, but the clock is past it.
+        Assert.False(wfc.DeadlineTask.IsCompleted);
+        Assert.True(wfc.IsTornDown);
+        Assert.Throws<OperationCanceledException>(wfc.ThrowIfTornDown);
+        Assert.True(wfc.IsDeadlineExceeded);
+
+        // Observing it fires the deadline, so every later observer agrees.
+        Assert.True(wfc.DeadlineTask.IsCompleted);
+        Assert.True(wfc.Token.IsCancellationRequested);
+    }
+
+    [Fact]
+    public async Task WorkflowCancellation_BlockingDeadlineTokenCallback_DoesNotBlockDeadline()
+    {
+        // The service client registers a callback on the deadline token that
+        // blocks. The SDK's own deadline work must still complete.
+        var tm = new TerminationManager();
+        using var release = new ManualResetEventSlim();
+        var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        wfc.DeadlineToken.Register(() => release.Wait());
+
+        try
+        {
+            await Task.Run(wfc.TriggerDeadline).NoHang();
+            await wfc.DeadlineTask.NoHang();
+            Assert.True(wfc.Token.IsCancellationRequested);
+            Assert.True(wfc.IsTornDown);
+        }
+        finally
+        {
+            release.Set();
+            wfc.Dispose();
+        }
+    }
+
+    [Fact]
+    public async Task Hydration_BlockingClientDeadlineCallback_DoesNotBlockTheWait()
+    {
+        // The client registers a blocking callback on the deadline token and
+        // never completes its request. The wait must still end at the deadline.
+        var tm = new TerminationManager();
+        using var release = new ManualResetEventSlim();
+        var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var never = new TaskCompletionSource<int>();
+
+        try
+        {
+            var wait = DurableFunction.WaitWithDeadlineAsync(never.Task, wfc);
+            // Registered after the wait started (as a client that registers once
+            // its request is under way), so it runs before anything the wait
+            // itself registered on this token.
+            wfc.DeadlineToken.Register(() => release.Wait());
+            wfc.TriggerDeadline();
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => wait.NoHang());
+        }
+        finally
+        {
+            release.Set();
+            wfc.Dispose();
+        }
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task WaitForCondition_DeadlinePassesDuringStateSerialization_RecordsNoOutcome(bool conditionMet)
+    {
+        // The state serializer runs after the deadline check that follows the
+        // condition check; the deadline passes while it runs (the serializer
+        // fires it, deterministically). No RETRY, SUCCEED or FAIL may be recorded.
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var recorder = new RecordingBatcher();
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), recorder.Batcher);
+
+        var op = ctx.WaitForConditionAsync<int>(
+            (state, _, _) => Task.FromResult(state + 1),
+            new WaitForConditionConfig<int>
+            {
+                InitialState = 0,
+                WaitStrategy = conditionMet
+                    ? WaitStrategy.FromDelegate<int>((_, _) => WaitDecision.Stop())
+                    : WaitStrategy.Fixed<int>(TimeSpan.FromSeconds(1)),
+                Serializer = new DeadlineTriggeringSerializer(wfc),
+            },
+            name: "w");
+
+        await Task.WhenAny(op, tm.TerminationTask).NoHang();
+        await recorder.Batcher.DrainAsync();
+
+        Assert.True(op.IsCanceled || op.IsFaulted, "The outcome must not be accepted after the deadline.");
+        if (op.IsFaulted) Assert.NotNull(op.Exception);
+        Assert.DoesNotContain(recorder.Flushed,
+            u => u.Action == OperationAction.FAIL || u.Action == OperationAction.RETRY || u.Action == OperationAction.SUCCEED);
+    }
+
+    // ── Copilot review round 4 ─────────────────────────────────────────
+
+    [Fact]
+    public async Task TriggerDeadline_DisposedAsSoonAsDeadlineTaskIsSignalled_StillCancelsDeadlineToken()
+    {
+        // A hydration wait ends on DeadlineTask, and the wrapper then disposes at
+        // once. Dispose at exactly that point (on the firing thread, right after
+        // DeadlineTask is signalled): the client's deadline token must still be
+        // cancelled, so a cooperative hydration request stops.
+        var wfc = new WorkflowCancellation(new TerminationManager(), ControlledDeadline);
+        var clientCancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        wfc.DeadlineToken.Register(() => clientCancelled.TrySetResult());
+        wfc.AfterDeadlineSignalledForTest = wfc.Dispose;
+
+        wfc.TriggerDeadline();
+
+        await clientCancelled.Task.NoHang();
+    }
+
+    [Theory]
+    [InlineData("step")]
+    [InlineData("child")]
+    public async Task StartCheckpointFinishesAfterDeadlineBeforeTimerRuns_UserCodeDoesNotStart(string kind)
+    {
+        // The awaited START checkpoint completes only after the clock has passed
+        // the deadline; the deadline timer has not run (the threadpool delayed
+        // it), so the workflow token is not cancelled yet.
+        var tm = new TerminationManager();
+        var (wfc, deadlinePassed) = DeadlineWithTimerNotRun(tm);
+        using var _ = wfc;
+        await using var batcher = new CheckpointBatcher("t0", async (_, _, _) =>
+        {
+            await deadlinePassed();
+            return "t1";
+        });
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), batcher);
+        var userCodeRan = false;
+
+        Task op = kind == "step"
+            ? ctx.StepAsync<int>(
+                async (_, _) => { userCodeRan = true; await Task.CompletedTask; return 1; },
+                name: "s",
+                config: new StepConfig { Semantics = StepSemantics.AtMostOncePerRetry })
+            : ctx.RunInChildContextAsync<int>(
+                (_, _) => { userCodeRan = true; return Task.FromResult(1); },
+                name: "c");
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => op.NoHang());
+        Assert.False(userCodeRan);
+    }
+
+    [Fact]
+    public async Task WaitForConditionReplay_StateRestoreCrossesDeadline_CheckDoesNotStart()
+    {
+        // READY replay: restoring the prior state runs the user serializer, which
+        // takes long enough to cross the deadline before its timer has run.
+        var tm = new TerminationManager();
+        var (wfc, deadlinePassed) = DeadlineWithTimerNotRun(tm);
+        using var _ = wfc;
+        var state = new ExecutionState();
+        state.LoadFromCheckpoint(new InitialExecutionState
+        {
+            Operations = new List<Operation>
+            {
+                new()
+                {
+                    Id = OperationIdGenerator.HashOperationId("1"),
+                    Type = OperationTypes.Step,
+                    SubType = OperationSubTypes.WaitForCondition,
+                    Status = OperationStatuses.Ready,
+                    Name = "w",
+                    StepDetails = new StepDetails { Result = "5", Attempt = 1 },
+                },
+            },
+        });
+        var ctx = new DurableContext(state, tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)));
+        var checkRan = false;
+
+        var op = ctx.WaitForConditionAsync<int>(
+            (s, _, _) => { checkRan = true; return Task.FromResult(s + 1); },
+            new WaitForConditionConfig<int>
+            {
+                InitialState = 0,
+                WaitStrategy = WaitStrategy.Fixed<int>(TimeSpan.FromSeconds(1)),
+                Serializer = new SlowDeserializer(() => deadlinePassed().GetAwaiter().GetResult()),
+            },
+            name: "w");
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => op.NoHang());
+        Assert.False(checkRan);
+    }
+
+    // ── Copilot review round 5 ─────────────────────────────────────────
+
+    [Fact]
+    public async Task Lmi_HydrationFaultsAfterDeadline_ThrowsTimeout()
+    {
+        // The request fires the deadline itself and then fails with a
+        // non-cancellation error, before the deadline wait can win the race.
+        WorkflowCancellation? wfc = null;
+        var client = new FakeServiceClient
+        {
+            OnGetExecutionState = _ =>
+            {
+                wfc!.TriggerDeadline();
+                throw new DurableExecutionException("request failed as it was cut off");
+            },
+        };
+        var options = new DurableInvocationOptions
+        {
+            IsMultiConcurrency = true,
+            OnWorkflowCancellationCreated = created => wfc = created,
+        };
+
+        var ex = await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                (input, _) => Task.FromResult(input.Id!),
+                Input("arn:hydrate-fault-after-deadline", nextMarker: "page-2"),
+                Context(ControlledDeadline),
+                client,
+                options).NoHang());
+
+        Assert.IsType<DurableExecutionException>(ex.InnerException);
+    }
+
+    /// <summary>Default serializer whose Deserialize first runs a blocking wait.</summary>
+    private sealed class SlowDeserializer : Amazon.Lambda.Core.ILambdaSerializer
+    {
+        private readonly DefaultLambdaJsonSerializer _inner = new();
+        private readonly Action _beforeDeserialize;
+        public SlowDeserializer(Action beforeDeserialize) => _beforeDeserialize = beforeDeserialize;
+        public T Deserialize<T>(Stream requestStream)
+        {
+            _beforeDeserialize();
+            return _inner.Deserialize<T>(requestStream);
+        }
+        public void Serialize<T>(T response, Stream responseStream) => _inner.Serialize(response, responseStream);
+    }
+
+    /// <summary>Default serializer whose Serialize fires the deadline first.</summary>
+    private sealed class DeadlineTriggeringSerializer : Amazon.Lambda.Core.ILambdaSerializer
+    {
+        private readonly DefaultLambdaJsonSerializer _inner = new();
+        private readonly WorkflowCancellation _wfc;
+        public DeadlineTriggeringSerializer(WorkflowCancellation wfc) => _wfc = wfc;
+        public T Deserialize<T>(Stream requestStream) => _inner.Deserialize<T>(requestStream);
+        public void Serialize<T>(T response, Stream responseStream)
+        {
+            _wfc.TriggerDeadline();
+            _inner.Serialize(response, responseStream);
+        }
+    }
+
+    [Fact]
+    public async Task RunAsync_WorkflowReturnsFallbackAfterDeadline_ReportsDeadline()
+    {
+        // The workflow returns a value after the deadline passed, before its timer
+        // ran. It wins the race against DeadlineTask; the deadline must still
+        // decide the outcome.
+        var tm = new TerminationManager();
+        var (wfc, deadlinePassed) = DeadlineWithTimerNotRun(tm);
+        using var _ = wfc;
+        var workflowRan = false;
+
+        var result = await DurableExecutionHandler.RunAsync<string>(
+            new ExecutionState(),
+            tm,
+            async () =>
+            {
+                workflowRan = true;
+                await deadlinePassed();
+                return "fallback";
+            },
+            wfc).NoHang();
+
+        Assert.True(workflowRan);
+        Assert.True(result.DeadlineExceeded);
+        Assert.Null(result.Result);
+    }
+
+    [Fact]
+    public async Task RunAsync_WorkflowStartedAfterTeardown_RunsNoWorkflowCode()
+    {
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm);
+        wfc.BeginTeardown();
+        var ran = false;
+
+        await DurableExecutionHandler.RunAsync<string>(
+            new ExecutionState(), tm,
+            () => { ran = true; return Task.FromResult("x"); },
+            wfc).NoHang();
+
+        Assert.False(ran);
+    }
+
+    [Theory]
+    [InlineData("step")]
+    [InlineData("child")]
+    [InlineData("condition")]
+    public async Task Lmi_BodyConvertsDeadlineCancellation_NoFailOrRetryCheckpoint(string kind)
+    {
+        // The user body catches its cancelled token and throws a different
+        // exception. That must not be recorded as a real failure.
+        var client = new FakeServiceClient();
+        var retry = new StepConfig { RetryStrategy = RetryStrategy.Exponential(maxAttempts: 3) };
+
+        async Task<string> Body(CancellationToken ct)
+        {
+            try { await Task.Delay(Timeout.Infinite, ct); }
+            catch (OperationCanceledException) { throw new InvalidOperationException("gave up"); }
+            return "unreachable";
+        }
+
+        Func<Payload, IDurableContext, Task<string>> workflow = kind switch
+        {
+            "step" => async (_, ctx) => await ctx.StepAsync<string>((_, ct) => Body(ct), name: "s", config: retry),
+            "child" => async (_, ctx) => await ctx.RunInChildContextAsync<string>((_, ct) => Body(ct), name: "c"),
+            _ => async (_, ctx) => (await ctx.WaitForConditionAsync<int>(
+                async (state, _, ct) => { await Body(ct); return state; },
+                new WaitForConditionConfig<int>
+                {
+                    InitialState = 0,
+                    WaitStrategy = WaitStrategy.Fixed<int>(TimeSpan.FromSeconds(1)),
+                },
+                name: "w")).ToString(),
+        };
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync(workflow, Input($"arn:convert-{kind}"),
+                Context(TimeSpan.FromMilliseconds(100)), client, LmiOptions).NoHang());
+
+        Assert.DoesNotContain(client.AllUpdates,
+            u => u.Action == OperationAction.FAIL || u.Action == OperationAction.RETRY);
+    }
+
+    [Theory]
+    [InlineData("step")]
+    [InlineData("child")]
+    [InlineData("condition")]
+    public async Task Operation_BodyConvertsDeadlineCancellation_EnqueuesNoFailOrRetry(string kind)
+    {
+        // Drives the operation directly, with nothing tearing the batcher down,
+        // so a FAIL/RETRY the operation enqueued would definitely be flushed.
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var recorder = new RecordingBatcher();
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), recorder.Batcher);
+        var retry = new StepConfig { RetryStrategy = RetryStrategy.Exponential(maxAttempts: 3) };
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        async Task<int> Body(CancellationToken ct)
+        {
+            entered.TrySetResult();
+            try { await Task.Delay(Timeout.Infinite, ct); }
+            catch (OperationCanceledException) { throw new InvalidOperationException("gave up"); }
+            return 0;
+        }
+
+        Task op = kind switch
+        {
+            "step" => ctx.StepAsync<int>((_, ct) => Body(ct), name: "s", config: retry),
+            "child" => ctx.RunInChildContextAsync<int>((_, ct) => Body(ct), name: "c"),
+            _ => ctx.WaitForConditionAsync<int>(
+                async (state, _, ct) => { await Body(ct); return state; },
+                new WaitForConditionConfig<int>
+                {
+                    InitialState = 0,
+                    WaitStrategy = WaitStrategy.Fixed<int>(TimeSpan.FromSeconds(1)),
+                },
+                name: "w"),
+        };
+
+        await entered.Task.NoHang();
+        wfc.TriggerDeadline();
+
+        // The operation must surface the error without recording a failure. A
+        // RETRY would suspend instead, so only wait for completion or suspension.
+        await Task.WhenAny(op, tm.TerminationTask).NoHang();
+        await recorder.Batcher.DrainAsync();
+
+        Assert.True(op.IsFaulted, "The operation should rethrow the body's exception.");
+        // Observe the fault: other tests count process-wide unobserved exceptions.
+        Assert.NotNull(op.Exception);
+        Assert.DoesNotContain(recorder.Flushed,
+            u => u.Action == OperationAction.FAIL || u.Action == OperationAction.RETRY);
+    }
+
+    [Fact]
+    public async Task StepAsync_TornDownWhileStartCheckpointIsAwaited_BodyDoesNotRun()
+    {
+        // AtMostOncePerRetry awaits its START checkpoint before the body. The
+        // workflow is torn down while that flush is in progress.
+        var flushEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseFlush = new TaskCompletionSource<string?>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm);
+        await using var batcher = new CheckpointBatcher("t0", (_, _, _) =>
+        {
+            flushEntered.TrySetResult();
+            return releaseFlush.Task;
+        });
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), batcher);
+        var bodyRan = false;
+
+        var step = ctx.StepAsync<int>(
+            async (_, _) => { bodyRan = true; await Task.CompletedTask; return 1; },
+            name: "s",
+            config: new StepConfig { Semantics = StepSemantics.AtMostOncePerRetry });
+
+        await flushEntered.Task.NoHang();
+        wfc.BeginTeardown();
+        releaseFlush.SetResult("t1");
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => step.NoHang());
+        Assert.False(bodyRan);
+    }
+
+    [Fact]
+    public async Task Lmi_DeadlinePassesDuringFinalDrain_ThrowsTimeout()
+    {
+        // The workflow finishes in time; its final checkpoint is still in flight
+        // when the deadline passes and completes inside the cleanup budget.
+        var client = new FakeServiceClient
+        {
+            OnCheckpoint = _ => Task.Delay(400),
+        };
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                async (input, ctx) =>
+                {
+                    // Fire-and-forget START plus a SUCCEED queued at the very end.
+                    ctx.StepAsync<string>(async (_, _) => { await Task.CompletedTask; return "a"; }, name: "a").Observe();
+                    await Task.Yield();
+                    return "done";
+                },
+                Input("arn:late-drain"),
+                Context(TimeSpan.FromMilliseconds(100)),
+                client,
+                new DurableInvocationOptions { IsMultiConcurrency = true, CleanupTimeout = TimeSpan.FromSeconds(5) })
+            .NoHang());
+    }
+
+    [Fact]
+    public async Task Lmi_HydrationIgnoringCancellation_IsStillBounded()
+    {
+        var never = new TaskCompletionSource();
+        var client = new FakeServiceClient { OnGetExecutionState = _ => never.Task };
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                (input, _) => Task.FromResult(input.Id!),
+                Input("arn:hydrate-ignores-token", nextMarker: "page-2"),
+                Context(TimeSpan.FromMilliseconds(100)),
+                client,
+                LmiOptions).NoHang());
+    }
+
+    [Fact]
+    public async Task CheckpointBatcher_AfterAbandon_SendsNoFurtherBatches()
+    {
+        // The worker must send two batches in one pass: with one operation per
+        // batch, reading "a" and "b" together flushes [a] mid-pass, then [b].
+        // The [a] flush ignores cancellation and stalls; the drain gives up and
+        // abandons the worker. Releasing [a] afterwards must not send [b].
+        var calls = 0;
+        var warmupEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseWarmup = new TaskCompletionSource<string?>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var stalledEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseStalled = new TaskCompletionSource<string?>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var batcher = new CheckpointBatcher("t0", (_, _, _) =>
+        {
+            switch (Interlocked.Increment(ref calls))
+            {
+                case 1: warmupEntered.TrySetResult(); return releaseWarmup.Task;
+                case 2: stalledEntered.TrySetResult(); return releaseStalled.Task;
+                default: return Task.FromResult<string?>("t-next");
+            }
+        }, new CheckpointBatcherConfig { MaxBatchOperations = 1 });
+
+        // Hold the worker in a warm-up flush so "a" and "b" queue up together.
+        batcher.EnqueueAsync(Update("w")).Observe();
+        await warmupEntered.Task.NoHang();
+        batcher.EnqueueAsync(Update("a")).Observe();
+        var second = batcher.EnqueueAsync(Update("b"));
+        releaseWarmup.SetResult("t1");
+
+        await stalledEntered.Task.NoHang();
+        Assert.False(await batcher.DrainAsync(TimeSpan.FromMilliseconds(50)));
+        releaseStalled.SetResult("t2");
+
+        await Assert.ThrowsAnyAsync<Exception>(() => second.NoHang());
+        Assert.Equal(2, Volatile.Read(ref calls));
+        await batcher.DisposeAsync();
+    }
+
+    // ── Copilot review round 2 ─────────────────────────────────────────
+
+    [Fact]
+    public async Task CheckpointBatcher_Abandon_DoesNotWaitForBlockingClientCallback()
+    {
+        // The in-flight flush registers a cancellation callback that blocks. The
+        // drain must still give up once its budget is spent.
+        using var releaseCallback = new ManualResetEventSlim();
+        var flushEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var never = new TaskCompletionSource<string?>();
+        var batcher = new CheckpointBatcher("t0", (_, _, ct) =>
+        {
+            ct.Register(() => releaseCallback.Wait());
+            flushEntered.TrySetResult();
+            return never.Task;
+        });
+
+        try
+        {
+            batcher.EnqueueAsync(Update("a")).Observe();
+            await flushEntered.Task.NoHang();
+
+            Assert.False(await batcher.DrainAsync(TimeSpan.FromMilliseconds(50)).NoHang());
+        }
+        finally
+        {
+            releaseCallback.Set();
+        }
+    }
+
+    [Fact]
+    public async Task CheckpointBatcher_Abandon_ReleasesPendingAwaiters()
+    {
+        // A flush that never returns would otherwise keep every waiting caller
+        // (in flight or queued) suspended forever.
+        var flushEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var never = new TaskCompletionSource<string?>();
+        var batcher = new CheckpointBatcher("t0", (_, _, _) =>
+        {
+            flushEntered.TrySetResult();
+            return never.Task;
+        });
+
+        var inFlight = batcher.EnqueueAsync(Update("a"));
+        await flushEntered.Task.NoHang();
+        var queued = batcher.EnqueueAsync(Update("b"));
+
+        Assert.False(await batcher.DrainAsync(TimeSpan.FromMilliseconds(50)));
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => inFlight.NoHang());
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => queued.NoHang());
+    }
+
+    [Fact]
+    public async Task StepAsync_WorkflowCancelledButLinkedTokenLagging_NoFailCheckpoint()
+    {
+        // Cancellation is delivered on another thread and stalls in a callback
+        // registered after the step's linked token, so the workflow token is
+        // cancelled while the linked one is not yet. The pre-body check then
+        // throws; that must count as cancellation, never as a step failure.
+        // A custom logger's BeginScope runs after the linked token is created and
+        // before the check, which is where the test arranges the lag.
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm);
+        var recorder = new RecordingBatcher();
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), recorder.Batcher);
+        using var releaseDelivery = new ManualResetEventSlim();
+        using var deliveryStalled = new ManualResetEventSlim();
+        ctx.ConfigureLogger(new LoggerConfig
+        {
+            CustomLogger = new ScopeHookLogger(() =>
+            {
+                wfc.Token.Register(() => { deliveryStalled.Set(); releaseDelivery.Wait(); });
+                // Suspension: TerminationManager delivers the cancel on the threadpool.
+                tm.Terminate(TerminationReason.WaitScheduled);
+                Assert.True(deliveryStalled.Wait(LambdaManagedInstancesTests.HangGuardTimeout));
+            }),
+        });
+        var bodyRan = false;
+
+        try
+        {
+            var step = ctx.StepAsync<int>(
+                async (_, _) => { bodyRan = true; await Task.CompletedTask; return 1; }, name: "s");
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => step.NoHang());
+        }
+        finally
+        {
+            releaseDelivery.Set();
+        }
+
+        await recorder.Batcher.DrainAsync();
+        Assert.False(bodyRan);
+        Assert.DoesNotContain(recorder.Flushed,
+            u => u.Action == OperationAction.FAIL || u.Action == OperationAction.RETRY);
+    }
+
+    [Fact]
+    public async Task ParallelAsync_BranchFailsWhileDeadlineDeliveryLags_NoParentCheckpoint()
+    {
+        // Deadline delivery stalls in a callback registered last, so it runs
+        // first and holds back every other callback, including the parallel's
+        // control token. The branches fail (converting the cancellation) before
+        // the control token is notified. The parent must not checkpoint an
+        // outcome that contains those failures.
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var recorder = new RecordingBatcher();
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), recorder.Batcher);
+        using var releaseDelivery = new ManualResetEventSlim();
+        var deliveryStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var entered = new CountdownEvent(2);
+
+        var branches = new Func<IDurableContext, CancellationToken, Task<int>>[2];
+        for (var i = 0; i < 2; i++)
+        {
+            branches[i] = async (_, _) =>
+            {
+                entered.Signal();
+                await deliveryStarted.Task;
+                throw new InvalidOperationException("gave up");
+            };
+        }
+
+        try
+        {
+            var run = ctx.ParallelAsync(branches, name: "fanout",
+                config: new ParallelConfig { CompletionConfig = CompletionConfig.AllCompleted() });
+            Assert.True(entered.Wait(LambdaManagedInstancesTests.HangGuardTimeout));
+            // Registered after the parallel's control token, so it runs before it.
+            wfc.Token.Register(() => { deliveryStarted.TrySetResult(); releaseDelivery.Wait(); });
+            wfc.TriggerDeadline();
+
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => run.NoHang());
+        }
+        finally
+        {
+            releaseDelivery.Set();
+        }
+
+        await recorder.Batcher.DrainAsync();
+        Assert.DoesNotContain(recorder.Flushed,
+            u => u.Type == OperationTypes.Context && u.SubType == OperationSubTypes.Parallel
+                 && (u.Action == OperationAction.SUCCEED || u.Action == OperationAction.FAIL));
+    }
+
+    [Fact]
+    public async Task WaitForCondition_CheckReturnsStateAfterDeadline_RecordsNoRetryOrFail()
+    {
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var recorder = new RecordingBatcher();
+        var ctx = new DurableContext(new ExecutionState(), tm, wfc, new OperationIdGenerator(),
+            "arn:test", Context(TimeSpan.FromMinutes(5)), recorder.Batcher);
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        // The check swallows the cancellation and returns a state.
+        var op = ctx.WaitForConditionAsync<int>(
+            async (state, _, ct) =>
+            {
+                entered.TrySetResult();
+                try { await Task.Delay(Timeout.Infinite, ct); }
+                catch (OperationCanceledException) { }
+                return state + 1;
+            },
+            new WaitForConditionConfig<int>
+            {
+                InitialState = 0,
+                WaitStrategy = WaitStrategy.Fixed<int>(TimeSpan.FromSeconds(1)),
+            },
+            name: "w");
+
+        await entered.Task.NoHang();
+        wfc.TriggerDeadline();
+        await Task.WhenAny(op, tm.TerminationTask).NoHang();
+        await recorder.Batcher.DrainAsync();
+
+        Assert.True(op.IsCanceled || op.IsFaulted, "The check outcome must not be accepted after the deadline.");
+        if (op.IsFaulted) Assert.NotNull(op.Exception);
+        Assert.DoesNotContain(recorder.Flushed,
+            u => u.Action == OperationAction.FAIL || u.Action == OperationAction.RETRY || u.Action == OperationAction.SUCCEED);
+    }
+
+    [Fact]
+    public async Task WorkflowCancellation_ThrowingDeadlineTokenCallback_IsContained()
+    {
+        // The deadline token is also given to the service client. A callback it
+        // registers that throws must not escape the deadline timer (which would
+        // crash the test host process, as it would an LMI environment).
+        var tm = new TerminationManager();
+        using var wfc = new WorkflowCancellation(tm, ControlledDeadline);
+        var laterCallbackRan = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        wfc.DeadlineToken.Register(() => laterCallbackRan.TrySetResult());
+        wfc.DeadlineToken.Register(() => throw new InvalidOperationException("client callback threw"));
+
+        // Firing must not throw here, on the caller's thread, either.
+        wfc.TriggerDeadline();
+        await wfc.DeadlineTask.NoHang();
+        await laterCallbackRan.Task.NoHang();
+        Assert.True(wfc.Token.IsCancellationRequested);
+    }
+
+    [Theory]
+    [InlineData(10_000, 2_000, 4_000, 8_000)]  // before the deadline: rest of it plus cleanup
+    [InlineData(10_000, 2_000, 11_500, 500)]   // 1.5s past the deadline: 0.5s of cleanup left
+    [InlineData(10_000, 2_000, 13_000, 0)]     // budget spent
+    public void DrainTimeout_IsTheRemainingCombinedBudget(int deadlineMs, int cleanupMs, int elapsedMs, int expectedMs)
+    {
+        var timeout = DurableFunction.ComputeDrainTimeout(
+            TimeSpan.FromMilliseconds(deadlineMs), TimeSpan.FromMilliseconds(cleanupMs), TimeSpan.FromMilliseconds(elapsedMs));
+        Assert.Equal(TimeSpan.FromMilliseconds(expectedMs), timeout);
+    }
+
+    [Fact]
+    public async Task Lmi_OutputSerializationCrossesDeadline_ThrowsTimeout()
+    {
+        var client = new FakeServiceClient();
+        var context = Context(TimeSpan.FromMilliseconds(100));
+        // Serializing the output takes longer than the time left. A sleep is used
+        // only to make the deadline pass; nothing asserts on elapsed time.
+        context.Serializer = new SlowSerializer(TimeSpan.FromMilliseconds(400));
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                (_, _) => Task.FromResult("done"),
+                Input("arn:slow-output"),
+                context,
+                client,
+                LmiOptions).NoHang());
+    }
+
+    /// <summary>Delegates to the default serializer; Serialize sleeps first.</summary>
+    private sealed class SlowSerializer : Amazon.Lambda.Core.ILambdaSerializer
+    {
+        private readonly DefaultLambdaJsonSerializer _inner = new();
+        private readonly TimeSpan _delay;
+        public SlowSerializer(TimeSpan delay) => _delay = delay;
+        public T Deserialize<T>(Stream requestStream) => _inner.Deserialize<T>(requestStream);
+        public void Serialize<T>(T response, Stream responseStream)
+        {
+            Thread.Sleep(_delay);
+            _inner.Serialize(response, responseStream);
+        }
+    }
+
+    /// <summary>A logger whose first BeginScope call runs a hook.</summary>
+    private sealed class ScopeHookLogger : Microsoft.Extensions.Logging.ILogger
+    {
+        private Action? _hook;
+        public ScopeHookLogger(Action hook) => _hook = hook;
+
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull
+        {
+            Interlocked.Exchange(ref _hook, null)?.Invoke();
+            return null;
+        }
+
+        public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => false;
+
+        public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
+            TState state, Exception? exception, Func<TState, Exception?, string> formatter) { }
+    }
+
+    private static SdkOperationUpdate Update(string id) => new()
+    {
+        Id = id,
+        Type = OperationTypes.Step,
+        Action = OperationAction.START,
+    };
+
+    [Fact]
+    public async Task Lmi_DeadlinePassesDuringHydration_ThrowsTimeout()
+    {
+        var client = new FakeServiceClient
+        {
+            OnGetExecutionState = ct => Task.Delay(Timeout.Infinite, ct),
+        };
+        var workflowRan = false;
+
+        var ex = await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                (input, _) => { workflowRan = true; return Task.FromResult(input.Id!); },
+                Input("arn:deadline-hydrate", nextMarker: "page-2"),
+                Context(TimeSpan.FromMilliseconds(100)),
+                client,
+                LmiOptions).NoHang());
+
+        Assert.IsAssignableFrom<OperationCanceledException>(ex.InnerException);
+        Assert.False(workflowRan);
+    }
+
+    [Fact]
+    public async Task Lmi_CheckpointCallStallsPastDeadline_CleanupIsBounded()
+    {
+        // The checkpoint call ignores its cancellation token and never returns.
+        // The invocation must still end instead of holding a processing loop.
+        var neverCompletes = new TaskCompletionSource();
+        var client = new FakeServiceClient { OnCheckpoint = _ => neverCompletes.Task };
+
+        await Assert.ThrowsAsync<TimeoutException>(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                async (_, ctx) => await ctx.StepAsync<string>(
+                    async (_, _) => { await Task.CompletedTask; return "done"; },
+                    name: "step"),
+                Input("arn:stalled-checkpoint"),
+                Context(TimeSpan.FromMilliseconds(100)),
+                client,
+                new DurableInvocationOptions { IsMultiConcurrency = true, CleanupTimeout = TimeSpan.FromMilliseconds(100) })
+            .NoHang());
+    }
+
+    [Fact]
+    public async Task Lmi_CompletesBeforeDeadline_Succeeds()
+    {
+        var client = new FakeServiceClient();
+
+        var output = await DurableFunction.WrapAsync<Payload, string>(
+            async (input, ctx) => await ctx.StepAsync<string>(
+                async (_, _) => { await Task.CompletedTask; return input.Id!; }, name: "step"),
+            Input("arn:in-time", id: "abc"),
+            Context(TimeSpan.FromMinutes(5)),
+            client,
+            LmiOptions);
+
+        Assert.Equal(InvocationStatus.Succeeded, output.Status);
+        Assert.Equal("\"abc\"", output.Result);
+    }
+
+    // ── Concurrent invocations ─────────────────────────────────────────
+
+    [Fact]
+    public async Task Lmi_ConcurrentInvocations_AreIsolated()
+    {
+        const int count = 8;
+        var client = new FakeServiceClient();
+        var handler = new SharedHandler(count);
+
+        // All invocations go through one handler instance, as they would in one
+        // LMI execution environment. Even-numbered ones suspend, odd ones finish.
+        var tasks = Enumerable.Range(0, count).Select(i => Task.Run(() =>
+            DurableFunction.WrapAsync<Payload, string>(
+                handler.RunAsync,
+                Input($"arn:exec-{i}", id: $"id-{i}", checkpointToken: $"initial-{i}"),
+                Context(TimeSpan.FromMinutes(5), requestId: $"req-{i}"),
+                client,
+                LmiOptions))).ToArray();
+
+        var outputs = await Task.WhenAll(tasks).NoHang();
+
+        for (var i = 0; i < count; i++)
+        {
+            var arn = $"arn:exec-{i}";
+
+            if (i % 2 == 0)
+            {
+                Assert.Equal(InvocationStatus.Pending, outputs[i].Status);
+            }
+            else
+            {
+                Assert.Equal(InvocationStatus.Succeeded, outputs[i].Status);
+                Assert.Equal($"\"id-{i}:id-{i}\"", outputs[i].Result);
+            }
+
+            // Each step saw only its own invocation's context.
+            Assert.Equal(arn, handler.ObservedArns[$"id-{i}"]);
+            Assert.Equal($"req-{i}", handler.ObservedRequestIds[$"id-{i}"]);
+
+            // Each execution's checkpoints carry only its own ids, payloads and
+            // tokens, and its token chain starts from its own initial token.
+            var calls = client.CallsFor(arn);
+            Assert.NotEmpty(calls);
+            Assert.Equal($"initial-{i}", calls[0].TokenIn);
+            for (var c = 1; c < calls.Count; c++)
+                Assert.Equal(calls[c - 1].TokenOut, calls[c].TokenIn);
+
+            var updates = calls.SelectMany(c => c.Updates).ToList();
+            Assert.All(updates, u => Assert.Contains(u.Id, new[] { IdAt(1), IdAt(2), IdAt(3) }));
+            Assert.All(updates.Where(u => u.Payload != null), u => Assert.Contains($"id-{i}", u.Payload));
+            Assert.Contains(updates, u => u.Id == IdAt(1) && u.Action == OperationAction.SUCCEED);
+        }
+    }
+
+    [Fact]
+    public async Task Lmi_PeerTimesOut_HealthyPeerIsUnaffected()
+    {
+        var client = new FakeServiceClient();
+        var peerFinished = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var healthyEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        CancellationToken healthyToken = default;
+
+        var healthy = Task.Run(() => DurableFunction.WrapAsync<Payload, string>(
+            async (input, ctx) => await ctx.StepAsync<string>(async (_, ct) =>
+            {
+                healthyToken = ct;
+                healthyEntered.TrySetResult();
+                // Keep running until the peer has timed out and been cleaned up.
+                await peerFinished.Task;
+                return input.Id!;
+            }, name: "healthy"),
+            Input("arn:healthy", id: "ok"),
+            Context(TimeSpan.FromMinutes(5)),
+            client,
+            LmiOptions));
+
+        await healthyEntered.Task.NoHang();
+
+        var timingOut = DurableFunction.WrapAsync<Payload, string>(
+            async (_, ctx) => await ctx.StepAsync<string>(async (_, ct) =>
+            {
+                await Task.Delay(Timeout.Infinite, ct);
+                return "unreachable";
+            }, name: "slow"),
+            Input("arn:timing-out"),
+            Context(TimeSpan.FromMilliseconds(100)),
+            client,
+            LmiOptions);
+
+        await Assert.ThrowsAsync<TimeoutException>(() => timingOut.NoHang());
+        Assert.False(healthyToken.IsCancellationRequested);
+
+        peerFinished.SetResult();
+        var output = await healthy.NoHang();
+
+        Assert.Equal(InvocationStatus.Succeeded, output.Status);
+        Assert.Equal("\"ok\"", output.Result);
+        Assert.Contains(client.CallsFor("arn:healthy").SelectMany(c => c.Updates),
+            u => u.Action == OperationAction.SUCCEED);
+    }
+
+    [Fact]
+    public async Task Lmi_ReplayWhilePeerIsLive_SkipsCompletedSteps()
+    {
+        var client = new FakeServiceClient();
+        var bothInside = new AsyncBarrier(2);
+        var replayBodyRuns = 0;
+        var freshBodyRuns = 0;
+
+        // The replaying execution already completed step 1; only step 2 runs.
+        var replay = Task.Run(() => DurableFunction.WrapAsync<Payload, string>(
+            async (_, ctx) =>
+            {
+                var first = await ctx.StepAsync<string>(async (_, _) =>
+                {
+                    Interlocked.Increment(ref replayBodyRuns);
+                    await Task.CompletedTask;
+                    return "recomputed";
+                }, name: "first");
+                var second = await ctx.StepAsync<string>(async (_, _) =>
+                {
+                    await bothInside.SignalAndWaitAsync();
+                    return "second";
+                }, name: "second");
+                return $"{first}+{second}";
+            },
+            Input("arn:replay", extraOperations: new[]
+            {
+                new Operation
+                {
+                    Id = IdAt(1), Type = OperationTypes.Step, Status = OperationStatuses.Succeeded,
+                    Name = "first", StepDetails = new StepDetails { Result = "\"cached\"" },
+                },
+            }),
+            Context(TimeSpan.FromMinutes(5)),
+            client,
+            LmiOptions));
+
+        var fresh = Task.Run(() => DurableFunction.WrapAsync<Payload, string>(
+            async (_, ctx) => await ctx.StepAsync<string>(async (_, _) =>
+            {
+                Interlocked.Increment(ref freshBodyRuns);
+                await bothInside.SignalAndWaitAsync();
+                return "fresh";
+            }, name: "first"),
+            Input("arn:fresh"),
+            Context(TimeSpan.FromMinutes(5)),
+            client,
+            LmiOptions));
+
+        var outputs = await Task.WhenAll(replay, fresh).NoHang();
+
+        Assert.Equal("\"cached+second\"", outputs[0].Result);
+        Assert.Equal("\"fresh\"", outputs[1].Result);
+        Assert.Equal(0, replayBodyRuns);
+        Assert.Equal(1, freshBodyRuns);
+        Assert.DoesNotContain(client.CallsFor("arn:replay").SelectMany(c => c.Updates), u => u.Id == IdAt(1));
+    }
+
+    // ── Helpers ─────────────────────────────────────────────────────────
+
+    /// <summary>The configured hang guard, for <see cref="HangGuardExtensions"/>.</summary>
+    internal static TimeSpan HangGuardTimeout => HangGuard;
+
+    /// <summary>
+    /// A deadline far enough away that its timer never fires during a test.
+    /// Tests fire it themselves with <see cref="WorkflowCancellation.TriggerDeadline"/>
+    /// once their setup is in place, so setup never races a real timer.
+    /// </summary>
+    private static readonly TimeSpan ControlledDeadline = TimeSpan.FromMinutes(10);
+
+    /// <summary>
+    /// A short deadline whose timer is never armed, plus a function that waits
+    /// until the clock has passed it: lets a workflow start, then finish after
+    /// the deadline, before anything fired it. Models the threadpool delaying
+    /// the timer callback. Only the outcome is asserted, never elapsed time.
+    /// </summary>
+    private static (WorkflowCancellation Wfc, Func<Task> DeadlinePassed) DeadlineWithTimerNotRun(TerminationManager tm)
+    {
+        var anchor = System.Diagnostics.Stopwatch.GetTimestamp();
+        var deadline = TimeSpan.FromMilliseconds(100);
+        var wfc = new WorkflowCancellation(tm, deadline, anchor, armTimer: false);
+        async Task DeadlinePassed()
+        {
+            while (System.Diagnostics.Stopwatch.GetElapsedTime(anchor) <= deadline)
+                await Task.Delay(10);
+        }
+        return (wfc, DeadlinePassed);
+    }
+
+    /// <summary>
+    /// A deadline the clock has already passed, whose timer has not run: models
+    /// the threadpool delaying the timer callback.
+    /// </summary>
+    private static WorkflowCancellation ExpiredWithTimerNotRun(TerminationManager tm) =>
+        new(tm, TimeSpan.FromSeconds(1),
+            anchorTimestamp: System.Diagnostics.Stopwatch.GetTimestamp() - 5 * System.Diagnostics.Stopwatch.Frequency,
+            armTimer: false);
+
+    /// <summary>
+    /// One handler instance shared by every concurrent invocation. Every step
+    /// waits until all invocations are inside a step, so they truly overlap.
+    /// </summary>
+    private sealed class SharedHandler
+    {
+        private readonly AsyncBarrier _allInside;
+        public ConcurrentDictionary<string, string> ObservedArns { get; } = new();
+        public ConcurrentDictionary<string, string> ObservedRequestIds { get; } = new();
+
+        public SharedHandler(int count) => _allInside = new AsyncBarrier(count);
+
+        public async Task<string> RunAsync(Payload input, IDurableContext ctx)
+        {
+            var first = await ctx.StepAsync<string>(async (_, _) =>
+            {
+                ObservedArns[input.Id!] = ctx.ExecutionContext.DurableExecutionArn;
+                ObservedRequestIds[input.Id!] = ctx.LambdaContext.AwsRequestId;
+                await _allInside.SignalAndWaitAsync();
+                return input.Id!;
+            }, name: "first");
+
+            var second = await ctx.StepAsync<string>(
+                async (_, _) => { await Task.Yield(); return input.Id!; }, name: "second");
+
+            var index = int.Parse(input.Id!.Substring("id-".Length));
+            if (index % 2 == 0)
+                await ctx.WaitAsync(TimeSpan.FromSeconds(30), name: "wait");
+
+            return $"{first}:{second}";
+        }
+    }
+
+    private sealed class AsyncBarrier
+    {
+        private int _remaining;
+        private readonly TaskCompletionSource _released = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public AsyncBarrier(int participants) => _remaining = participants;
+
+        public Task SignalAndWaitAsync()
+        {
+            if (Interlocked.Decrement(ref _remaining) == 0)
+                _released.TrySetResult();
+            return _released.Task;
+        }
+    }
+
+    public class Payload
+    {
+        public string? Id { get; set; }
+    }
+
+    private static TestLambdaContext Context(TimeSpan remainingTime, string requestId = "req") => new()
+    {
+        Serializer = new DefaultLambdaJsonSerializer(),
+        RemainingTime = remainingTime,
+        AwsRequestId = requestId,
+    };
+
+    private static DurableExecutionInvocationInput Input(
+        string arn,
+        string id = "id",
+        string checkpointToken = "initial",
+        string? nextMarker = null,
+        IEnumerable<Operation>? extraOperations = null)
+    {
+        var operations = new List<Operation>
+        {
+            new()
+            {
+                Id = "exec-0",
+                Type = OperationTypes.Execution,
+                Status = OperationStatuses.Started,
+                ExecutionDetails = new ExecutionDetails { InputPayload = $"{{\"Id\":\"{id}\"}}" },
+            },
+        };
+        if (extraOperations != null) operations.AddRange(extraOperations);
+
+        return new DurableExecutionInvocationInput
+        {
+            DurableExecutionArn = arn,
+            CheckpointToken = checkpointToken,
+            InitialExecutionState = new InitialExecutionState { Operations = operations, NextMarker = nextMarker },
+        };
+    }
+
+    private sealed record CheckpointCall(string Arn, string? TokenIn, string TokenOut, IReadOnlyList<SdkOperationUpdate> Updates);
+
+    /// <summary>
+    /// In-memory durable service shared by concurrent invocations. Issues a new
+    /// checkpoint token per call, scoped to the execution ARN.
+    /// </summary>
+    private sealed class FakeServiceClient : IDurableServiceClient
+    {
+        private readonly ConcurrentQueue<CheckpointCall> _calls = new();
+        private readonly ConcurrentQueue<CancellationToken> _hydrationTokens = new();
+        private int _tokenCounter;
+
+        public Func<CancellationToken, Task>? OnCheckpoint { get; init; }
+        public Func<CancellationToken, Task>? OnGetExecutionState { get; init; }
+
+        public IReadOnlyList<CheckpointCall> Calls => _calls.ToArray();
+        public IReadOnlyList<CancellationToken> HydrationTokens => _hydrationTokens.ToArray();
+        public IEnumerable<SdkOperationUpdate> AllUpdates => Calls.SelectMany(c => c.Updates);
+        public List<CheckpointCall> CallsFor(string arn) => Calls.Where(c => c.Arn == arn).ToList();
+
+        public async Task<string?> CheckpointAsync(
+            string durableExecutionArn,
+            string? checkpointToken,
+            IReadOnlyList<SdkOperationUpdate> pendingOperations,
+            Action<IReadOnlyList<Operation>>? onNewOperations = null,
+            CancellationToken cancellationToken = default)
+        {
+            if (OnCheckpoint != null) await OnCheckpoint(cancellationToken);
+
+            var tokenOut = $"{durableExecutionArn}#{Interlocked.Increment(ref _tokenCounter)}";
+            _calls.Enqueue(new CheckpointCall(durableExecutionArn, checkpointToken, tokenOut, pendingOperations.ToArray()));
+            return tokenOut;
+        }
+
+        public async Task<(List<Operation> Operations, string? NextMarker)> GetExecutionStateAsync(
+            string durableExecutionArn,
+            string? checkpointToken,
+            string marker,
+            CancellationToken cancellationToken = default)
+        {
+            _hydrationTokens.Enqueue(cancellationToken);
+            if (OnGetExecutionState != null) await OnGetExecutionState(cancellationToken);
+            return (new List<Operation>(), null);
+        }
+    }
+}
+
+/// <summary>
+/// Guards a test against hanging. Fails with an xUnit exception rather than
+/// <see cref="TimeoutException"/>: the deadline tests assert on
+/// <see cref="TimeoutException"/>, so <c>Task.WaitAsync(TimeSpan)</c> would turn
+/// a hang into a false pass.
+/// </summary>
+internal static class HangGuardExtensions
+{
+    /// <summary>
+    /// Observes a task the test deliberately does not await, so its fault never
+    /// reaches TaskScheduler.UnobservedTaskException. Other tests in this
+    /// assembly count that process-wide event.
+    /// </summary>
+    public static void Observe(this Task task) =>
+        task.ContinueWith(static t => _ = t.Exception, CancellationToken.None,
+            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
+
+    public static async Task NoHang(this Task task)
+    {
+        if (await Task.WhenAny(task, Task.Delay(LambdaManagedInstancesTests.HangGuardTimeout)) != task)
+            throw new Xunit.Sdk.XunitException("The operation did not complete within the hang guard; it appears to hang.");
+        await task;
+    }
+
+    public static async Task<T> NoHang<T>(this Task<T> task)
+    {
+        await ((Task)task).NoHang();
+        return await task;
+    }
+}