diff --git a/CHANGELOG.md b/CHANGELOG.md
index cac94858..901f5317 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -104,6 +104,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `durabletaskscheduler` token scopes now trim whitespace and all trailing slashes from `Options.ResourceID`. Previously only one trailing slash was removed and untrimmed values produced a malformed scope.
 - The owned DTS management client now observes both unary and streaming RPC outcomes for channel recreation. Caller cancellation, caller deadlines, and expected instance-wait deadlines are neutral and no longer reset or poison an existing transport-failure streak.
 - `TaskHubGrpcClient.ScheduleNewOrchestration` now sends the caller's W3C trace context on `CreateInstanceRequest`, so the service can parent the `create_orchestration` span to it. Previously every orchestration scheduled over gRPC or the Durable Task Scheduler began a disconnected root trace instead of joining the caller's trace.
+- `ContinueAsNew` with `task.WithKeepUnprocessedEvents` no longer drops an external event that arrives in the same work item after the orchestrator returns while a `WaitForSingleEvent` for that name is still pending, such as the loser of a `WhenAny` with a timer. Previously the event completed the unreachable wait and was missing from the next execution. Applications that re-sent such events to work around the loss can now see both copies. A failed or terminated completion no longer carries events forward when `ContinueAsNew` was called earlier in the turn.
+- Rejected rewind histories now identify the Go SDK's history-rewrite limitation instead of implying that DTS does not support rewind. The shared unsupported-feature error message is neutral about the source of the limitation; `errors.Is(err, api.ErrFeatureNotSupported)` and rewind safety checks are unchanged.
 
 ## [v0.6.0] - 2025-02-05
 
diff --git a/README.md b/README.md
index c0207767..7c26fc34 100644
--- a/README.md
+++ b/README.md
@@ -270,6 +270,10 @@ Two rules control the order:
 - If more than one live wait uses the same event name, the newest wait gets the next event. This rule is last in, first out. It agrees with the Durable Task .NET replay contract.
 - If an event arrives before any wait, DTS keeps the event. The orchestration then reads these events in arrival order. This rule is first in, first out.
 
+A wait stays live until it gets an event, times out, or is canceled. This is true even if the wait lost a `WhenAny` race. A live wait that no code awaits still takes the next event with its name. To keep the event for later code, cancel the losing wait with `ctx.WithCancel`. You can also read events with `task.NewEventChannel` and `ctx.Select`. A channel keeps an event until a `Select` case receives it.
+
+With `task.WithKeepUnprocessedEvents`, `ContinueAsNew` moves the undelivered events to the next execution. This includes an event that arrives in the same work item after the orchestrator function returns, even if a wait for that name is still live.
+
 Full sample: [samples/externalevents](./samples/externalevents).
 
 ### Pattern 4: Retries and failure handling
diff --git a/api/management.go b/api/management.go
index 312f4541..0d070731 100644
--- a/api/management.go
+++ b/api/management.go
@@ -17,7 +17,9 @@ const (
 	DefaultPurgePollInterval     = 100 * time.Millisecond
 )
 
-var ErrFeatureNotSupported = errors.New("feature is not supported by Durable Task Scheduler")
+// ErrFeatureNotSupported identifies a limitation in the SDK or service.
+// The surrounding error describes which operation or history shape is unsupported.
+var ErrFeatureNotSupported = errors.New("feature is not supported")
 
 // OrchestrationQuery describes a bounded query for orchestration instances.
 // ContinuationToken values are opaque and must only be reused with the same query.
diff --git a/client/grpc_errors_test.go b/client/grpc_errors_test.go
index 48e26a2b..43638550 100644
--- a/client/grpc_errors_test.go
+++ b/client/grpc_errors_test.go
@@ -42,6 +42,23 @@ func TestClientRPCErrorMappingsPreserveStatus(t *testing.T) {
 	}
 }
 
+func TestClientRPCUnsupportedFeaturePreservesServiceReason(t *testing.T) {
+	const reason = "server does not expose QueryInstances"
+	err := clientRPCError(context.Background(), "query instances", status.Error(codes.Unimplemented, reason))
+	if !errors.Is(err, api.ErrFeatureNotSupported) {
+		t.Fatalf("error %v does not match ErrFeatureNotSupported", err)
+	}
+	if status.Code(err) != codes.Unimplemented {
+		t.Fatalf("status.Code() = %v, want Unimplemented", status.Code(err))
+	}
+	if !strings.Contains(err.Error(), reason) {
+		t.Fatalf("error lost the service's reason: %v", err)
+	}
+	if strings.Contains(err.Error(), "not supported by Durable Task Scheduler") {
+		t.Fatalf("the shared error category should not attribute the limitation to DTS: %v", err)
+	}
+}
+
 func TestClientRPCErrorPrefersCallerContext(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
diff --git a/client/grpc_worker_rewind_test.go b/client/grpc_worker_rewind_test.go
index b30e84aa..fd2a9107 100644
--- a/client/grpc_worker_rewind_test.go
+++ b/client/grpc_worker_rewind_test.go
@@ -127,5 +127,6 @@ func TestWorkerRewindRejectsHandledFailureWithoutReplacement(t *testing.T) {
 	require.Nil(t, response.Actions[0].GetRewindOrchestration())
 	completion := response.Actions[0].GetCompleteOrchestration()
 	require.Equal(t, api.RUNTIME_STATUS_FAILED, completion.GetOrchestrationStatus())
-	require.Contains(t, completion.GetFailureDetails().GetErrorMessage(), "rewind cannot remove an operation failure")
+	require.Contains(t, completion.GetFailureDetails().GetErrorMessage(), "Go SDK cannot safely rewind this history")
+	require.NotContains(t, completion.GetFailureDetails().GetErrorMessage(), "not supported by Durable Task Scheduler")
 }
diff --git a/durabletaskscheduler/README.md b/durabletaskscheduler/README.md
index 36716fd3..0da4a562 100644
--- a/durabletaskscheduler/README.md
+++ b/durabletaskscheduler/README.md
@@ -267,6 +267,8 @@ unsupported history in its logs without emitting replacement history; the
 instance may retain its original failure metadata. Enqueue success does not
 imply that this worker-side validation succeeded. This guard is not a
 general proof of replay safety and does not infer whether a failure was handled.
+These history rejections describe a Go SDK limitation, not a lack of rewind
+support in DTS. They retain the `api.ErrFeatureNotSupported` error category.
 Rewind does not repair unchanged failing application code. Replacement history
 uses the normal large-payload handling and orchestration response-size bound;
 oversized responses fail explicitly rather than being silently truncated.
diff --git a/task/continue_as_new_events_test.go b/task/continue_as_new_events_test.go
new file mode 100644
index 00000000..b6d40743
--- /dev/null
+++ b/task/continue_as_new_events_test.go
@@ -0,0 +1,400 @@
+package task
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/microsoft/durabletask-go/api"
+	"github.com/microsoft/durabletask-go/internal/helpers"
+	"github.com/microsoft/durabletask-go/internal/protos"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+const (
+	continueAsNewEventsName     = "continue-as-new-events"
+	continueAsNewEventsInstance = api.InstanceID("continue-as-new-events-instance")
+)
+
+// timerWinsThenContinuesAsNew returns an orchestrator whose first generation
+// leaves the given number of "work" waits pending, continues as new when its
+// timer wins, and whose next generation returns the first "work" value.
+func timerWinsThenContinuesAsNew(waits int, keep bool) Orchestrator {
+	return func(ctx *OrchestrationContext) (any, error) {
+		var generation int
+		if err := ctx.GetInput(&generation); err != nil {
+			return nil, err
+		}
+		if generation > 0 {
+			var value int
+			err := ctx.WaitForSingleEvent("work", -1).Await(&value)
+			return value, err
+		}
+		tasks := make([]Task, 0, waits+1)
+		for range waits {
+			tasks = append(tasks, ctx.WaitForSingleEvent("work", -1))
+		}
+		timer := ctx.CreateTimer(time.Hour)
+		if ctx.WhenAny(append(tasks, timer)...) != timer {
+			return nil, errors.New("an event won before the timer")
+		}
+		var options []ContinueAsNewOption
+		if keep {
+			options = append(options, WithKeepUnprocessedEvents())
+		}
+		ctx.ContinueAsNew(generation+1, options...)
+		return nil, nil
+	}
+}
+
+func continueAsNewEventsRegistry(t *testing.T, orchestrator Orchestrator) *TaskRegistry {
+	t.Helper()
+	registry := NewTaskRegistry()
+	require.NoError(t, registry.AddOrchestratorN(continueAsNewEventsName, orchestrator))
+	return registry
+}
+
+func continueAsNewEventsStarted(input *wrapperspb.StringValue) []*protos.HistoryEvent {
+	return []*protos.HistoryEvent{
+		helpers.NewOrchestratorStartedEvent(),
+		helpers.NewExecutionStartedEvent(
+			continueAsNewEventsName,
+			string(continueAsNewEventsInstance),
+			input,
+			nil,
+			nil,
+			nil,
+		),
+	}
+}
+
+// afterFirstTimer runs a first turn that must only schedule one timer. It
+// returns the committed history and the event that fires that timer.
+func afterFirstTimer(t *testing.T, registry *TaskRegistry) ([]*protos.HistoryEvent, *protos.HistoryEvent) {
+	t.Helper()
+	history := continueAsNewEventsStarted(nil)
+	response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, nil, history...)
+	require.Len(t, response.Actions, 1)
+	timer := response.Actions[0].GetCreateTimer()
+	require.NotNil(t, timer, "the first turn must only schedule the timer")
+	timerID := response.Actions[0].Id
+	return append(history, helpers.NewTimerCreatedEvent(timerID, timer.GetFireAt())),
+		helpers.NewTimerFiredEvent(timerID, timer.GetFireAt(), nil)
+}
+
+func executeContinueAsNewTurn(
+	t *testing.T,
+	registry *TaskRegistry,
+	options OrchestrationOptions,
+	oldEvents []*protos.HistoryEvent,
+	newEvents ...*protos.HistoryEvent,
+) *protos.OrchestratorResponse {
+	t.Helper()
+	result, err := NewTaskExecutor(registry, WithOrchestrationOptions(options)).ExecuteOrchestrator(
+		context.Background(),
+		continueAsNewEventsInstance,
+		oldEvents,
+		newEvents,
+		supportedEntityParameters(),
+	)
+	require.NoError(t, err)
+	return result.Response
+}
+
+func raised(name string, value int) *protos.HistoryEvent {
+	return helpers.NewEventRaisedEvent(name, wrapperspb.String(strconv.Itoa(value)))
+}
+
+// requireContinuedAsNewCarryover asserts a CONTINUED_AS_NEW completion whose
+// carryover holds exactly want, formatted as name=payload, in order.
+func requireContinuedAsNewCarryover(
+	t *testing.T,
+	response *protos.OrchestratorResponse,
+	want ...string,
+) *protos.CompleteOrchestrationAction {
+	t.Helper()
+	completed := completionAction(t, response)
+	require.Equal(
+		t,
+		protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW,
+		completed.GetOrchestrationStatus(),
+	)
+	var carryover []string
+	for _, event := range completed.GetCarryoverEvents() {
+		eventRaised := event.GetEventRaised()
+		carryover = append(carryover, eventRaised.GetName()+"="+eventRaised.GetInput().GetValue())
+	}
+	require.Equal(t, want, carryover)
+	return completed
+}
+
+// TestContinueAsNewRetainsEventRaisedAfterTimerWins covers an event that is in
+// the same work item as the winning timer while a WaitForSingleEvent loser is
+// still pending. The finalized generation can never run that wait again.
+func TestContinueAsNewRetainsEventRaisedAfterTimerWins(t *testing.T) {
+	for _, keep := range []bool{true, false} {
+		t.Run(fmt.Sprintf("keep=%t", keep), func(t *testing.T) {
+			registry := continueAsNewEventsRegistry(t, timerWinsThenContinuesAsNew(1, keep))
+			oldEvents, timerFired := afterFirstTimer(t, registry)
+			response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, oldEvents,
+				helpers.NewOrchestratorStartedEvent(),
+				timerFired,
+				raised("work", 1),
+			)
+			if !keep {
+				requireContinuedAsNewCarryover(t, response)
+				return
+			}
+			completed := requireContinuedAsNewCarryover(t, response, "work=1")
+
+			nextGeneration := append(
+				continueAsNewEventsStarted(completed.GetResult()),
+				completed.GetCarryoverEvents()...,
+			)
+			next := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, nil, nextGeneration...)
+			require.Equal(t, "1", completionResult(t, next))
+		})
+	}
+}
+
+func TestContinueAsNewCarryoverKeepsArrivalOrderAfterTimerWins(t *testing.T) {
+	registry := continueAsNewEventsRegistry(t, timerWinsThenContinuesAsNew(2, true))
+	oldEvents, timerFired := afterFirstTimer(t, registry)
+	response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, oldEvents,
+		helpers.NewOrchestratorStartedEvent(),
+		raised("other", 0),
+		timerFired,
+		raised("work", 1),
+		raised("other", 2),
+		raised("WORK", 3),
+		raised("work", 4),
+	)
+	requireContinuedAsNewCarryover(t, response, "other=0", "work=1", "other=2", "WORK=3", "work=4")
+}
+
+// TestContinueAsNewFinalizationBoundary verifies that events keep their normal
+// delivery until a continue-as-new completion is finalized.
+func TestContinueAsNewFinalizationBoundary(t *testing.T) {
+	t.Run("event consumed before the timer is not carried over", func(t *testing.T) {
+		registry := continueAsNewEventsRegistry(t, func(ctx *OrchestrationContext) (any, error) {
+			pending := ctx.WaitForSingleEvent("work", -1)
+			if ctx.WhenAny(pending, ctx.CreateTimer(time.Hour)) != pending {
+				return nil, errors.New("the timer won")
+			}
+			var value int
+			if err := pending.Await(&value); err != nil {
+				return nil, err
+			}
+			ctx.ContinueAsNew(value, WithKeepUnprocessedEvents())
+			return nil, nil
+		})
+		oldEvents, timerFired := afterFirstTimer(t, registry)
+		response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, oldEvents,
+			helpers.NewOrchestratorStartedEvent(),
+			raised("work", 1),
+			raised("work", 2),
+			timerFired,
+		)
+		completed := requireContinuedAsNewCarryover(t, response, "work=2")
+		require.Equal(t, "1", completed.GetResult().GetValue())
+	})
+
+	t.Run("continue-as-new intent does not stop delivery", func(t *testing.T) {
+		registry := continueAsNewEventsRegistry(t, func(ctx *OrchestrationContext) (any, error) {
+			ctx.ContinueAsNew(0, WithKeepUnprocessedEvents())
+			var value int
+			if err := ctx.WaitForSingleEvent("work", -1).Await(&value); err != nil {
+				return nil, err
+			}
+			ctx.ContinueAsNew(value, WithKeepUnprocessedEvents())
+			return nil, nil
+		})
+		newEvents := append(continueAsNewEventsStarted(nil), raised("work", 1), raised("work", 2))
+		response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, nil, newEvents...)
+		completed := requireContinuedAsNewCarryover(t, response, "work=2")
+		require.Equal(t, "1", completed.GetResult().GetValue())
+	})
+}
+
+// TestContinueAsNewCarryoverForEventWaitPatterns covers the ways an event
+// receive can still be outstanding when a timer wins.
+func TestContinueAsNewCarryoverForEventWaitPatterns(t *testing.T) {
+	tests := []struct {
+		name         string
+		orchestrator Orchestrator
+	}{
+		{
+			name: "blocked event channel receive",
+			orchestrator: func(ctx *OrchestrationContext) (any, error) {
+				channel := NewEventChannel[int](ctx, "work")
+				ctx.Go(func(ctx *OrchestrationContext) {
+					_, _ = channel.ReceiveErr(ctx)
+				})
+				if err := ctx.CreateTimer(time.Hour).Await(nil); err != nil {
+					return nil, err
+				}
+				ctx.ContinueAsNew(1, WithKeepUnprocessedEvents())
+				return nil, nil
+			},
+		},
+		{
+			name: "canceled wait scope",
+			orchestrator: func(ctx *OrchestrationContext) (any, error) {
+				waitCtx, cancel := ctx.WithCancel()
+				pending := waitCtx.WaitForSingleEvent("work", -1)
+				timer := ctx.CreateTimer(time.Hour)
+				if ctx.WhenAny(pending, timer) != timer {
+					return nil, errors.New("an event won before the timer")
+				}
+				cancel()
+				ctx.ContinueAsNew(1, WithKeepUnprocessedEvents())
+				return nil, nil
+			},
+		},
+		{
+			name: "event channel select",
+			orchestrator: func(ctx *OrchestrationContext) (any, error) {
+				timerWon := false
+				ctx.Select(
+					OnTask(ctx.CreateTimer(time.Hour), func(Task) { timerWon = true }),
+					OnEvent(NewEventChannel[int](ctx, "work"), func(int) {}),
+				)
+				if !timerWon {
+					return nil, errors.New("an event won before the timer")
+				}
+				ctx.ContinueAsNew(1, WithKeepUnprocessedEvents())
+				return nil, nil
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			registry := continueAsNewEventsRegistry(t, test.orchestrator)
+			oldEvents, timerFired := afterFirstTimer(t, registry)
+			response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, oldEvents,
+				helpers.NewOrchestratorStartedEvent(),
+				timerFired,
+				raised("work", 1),
+			)
+			requireContinuedAsNewCarryover(t, response, "work=1")
+		})
+	}
+}
+
+// TestContinueAsNewIntentDoesNotAddCarryoverToOtherOutcomes verifies that only
+// an actual CONTINUED_AS_NEW completion carries events forward.
+func TestContinueAsNewIntentDoesNotAddCarryoverToOtherOutcomes(t *testing.T) {
+	tests := []struct {
+		name         string
+		orchestrator Orchestrator
+		trailing     []*protos.HistoryEvent
+		status       protos.OrchestrationStatus
+	}{
+		{
+			name: "orchestrator error",
+			orchestrator: func(ctx *OrchestrationContext) (any, error) {
+				ctx.ContinueAsNew(1, WithKeepUnprocessedEvents())
+				return nil, errors.New("orchestrator failed")
+			},
+			status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED,
+		},
+		{
+			name: "input serialization error",
+			orchestrator: func(ctx *OrchestrationContext) (any, error) {
+				ctx.ContinueAsNew(make(chan int), WithKeepUnprocessedEvents())
+				return nil, nil
+			},
+			status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED,
+		},
+		{
+			name: "termination",
+			orchestrator: func(ctx *OrchestrationContext) (any, error) {
+				ctx.ContinueAsNew(1, WithKeepUnprocessedEvents())
+				return nil, ctx.WaitForSingleEvent("never", -1).Await(nil)
+			},
+			trailing: []*protos.HistoryEvent{
+				helpers.NewExecutionTerminatedEvent(wrapperspb.String(`"stopped"`), false),
+			},
+			status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_TERMINATED,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			registry := continueAsNewEventsRegistry(t, test.orchestrator)
+			newEvents := append(continueAsNewEventsStarted(nil), raised("other", 1))
+			newEvents = append(newEvents, test.trailing...)
+			response := executeContinueAsNewTurn(t, registry, OrchestrationOptions{}, nil, newEvents...)
+			completed := completionAction(t, response)
+			require.Equal(t, test.status, completed.GetOrchestrationStatus())
+			require.Empty(t, completed.GetCarryoverEvents())
+		})
+	}
+}
+
+// TestContinueAsNewRetainsTrailingEventsAcrossTurnCapsAndSuspension covers
+// trailing events left unread by MaxEventsPerTurn and events buffered while
+// suspended that are drained after the timer when the orchestration resumes.
+func TestContinueAsNewRetainsTrailingEventsAcrossTurnCapsAndSuspension(t *testing.T) {
+	registry := continueAsNewEventsRegistry(t, timerWinsThenContinuesAsNew(1, true))
+	oldEvents, timerFired := afterFirstTimer(t, registry)
+	trailing := []*protos.HistoryEvent{raised("work", 1), raised("other", 2), raised("work", 3)}
+	tests := []struct {
+		name      string
+		options   OrchestrationOptions
+		oldEvents []*protos.HistoryEvent
+		newEvents []*protos.HistoryEvent
+	}{
+		{
+			name:      "turn cap after a trailing event",
+			options:   OrchestrationOptions{MaxEventsPerTurn: 2},
+			oldEvents: oldEvents,
+			newEvents: append([]*protos.HistoryEvent{helpers.NewOrchestratorStartedEvent(), timerFired}, trailing...),
+		},
+		{
+			name:      "turn cap before trailing events",
+			options:   OrchestrationOptions{MaxEventsPerTurn: 1},
+			oldEvents: oldEvents,
+			newEvents: append([]*protos.HistoryEvent{helpers.NewOrchestratorStartedEvent(), timerFired}, trailing...),
+		},
+		{
+			name:      "suspended and resumed in one work item",
+			oldEvents: oldEvents,
+			newEvents: []*protos.HistoryEvent{
+				helpers.NewOrchestratorStartedEvent(),
+				helpers.NewSuspendOrchestrationEvent("hold"),
+				timerFired,
+				trailing[0],
+				trailing[1],
+				helpers.NewResumeOrchestrationEvent("go"),
+				trailing[2],
+			},
+		},
+		{
+			name: "suspended history drained on resume",
+			oldEvents: append(slices.Clone(oldEvents),
+				helpers.NewOrchestratorStartedEvent(),
+				helpers.NewSuspendOrchestrationEvent("hold"),
+				timerFired,
+				trailing[0],
+				trailing[1],
+			),
+			newEvents: []*protos.HistoryEvent{
+				helpers.NewOrchestratorStartedEvent(),
+				helpers.NewResumeOrchestrationEvent("go"),
+				trailing[2],
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			response := executeContinueAsNewTurn(t, registry, test.options, test.oldEvents, test.newEvents...)
+			requireContinuedAsNewCarryover(t, response, "work=1", "other=2", "work=3")
+			require.Nil(t, response.NumEventsProcessed)
+		})
+	}
+}
diff --git a/task/orchestrator.go b/task/orchestrator.go
index 2727fa2b..83e43d86 100644
--- a/task/orchestrator.go
+++ b/task/orchestrator.go
@@ -84,6 +84,9 @@ type OrchestrationContext struct {
 	eventChannels              map[string]any
 	eventWaiters               map[string]map[*coroutine]struct{}
 	saveBufferedExternalEvents bool
+	// continueAsNewFinalized is true while a successfully built continue-as-new
+	// completion is this execution's pending terminal action.
+	continueAsNewFinalized bool
 
 	criticalSectionID               string
 	criticalSectionLocks            []string
@@ -119,8 +122,11 @@ type SubOrchestratorOption func(*callSubOrchestratorOptions, api.DataConverter)
 // ContinueAsNewOption is a functional option type for the ContinueAsNew orchestrator method.
 type ContinueAsNewOption func(*OrchestrationContext)
 
-// WithKeepUnprocessedEvents returns a ContinueAsNewOptions struct that instructs the
-// runtime to carry forward any unprocessed external events to the new instance.
+// WithKeepUnprocessedEvents instructs the runtime to carry forward external
+// events that this execution did not deliver. That includes events that arrive
+// after the orchestrator function returns, even if a wait for the same event
+// name is still pending. An event that completed a wait before the function
+// returned was delivered, even if the wait was never awaited.
 func WithKeepUnprocessedEvents() ContinueAsNewOption {
 	return func(ctx *OrchestrationContext) {
 		ctx.saveBufferedExternalEvents = true
@@ -348,6 +354,7 @@ func (ctx *OrchestrationContext) start() (actions []*protos.OrchestratorAction)
 	ctx.sequenceNumber = 0
 	ctx.newGuidCounter = 0
 	ctx.pendingActions = make(map[int32]*protos.OrchestratorAction)
+	ctx.continueAsNewFinalized = false
 	ctx.pendingTasks = make(map[int32]*completableTask)
 	ctx.pendingEntityTasks = make(map[string]*completableTask)
 	ctx.clearCriticalSection()
@@ -1377,7 +1384,10 @@ func (ctx *OrchestrationContext) onTimerFired(tf *protos.TimerFiredEvent) error
 func (ctx *OrchestrationContext) onExternalEventRaised(e *protos.HistoryEvent) error {
 	er := e.GetEventRaised()
 	key := strings.ToUpper(er.GetName())
-	if pendingTasks, ok := ctx.pendingExternalEventTasks[key]; ok && pendingTasks.Len() > 0 {
+	// Once continue-as-new is finalized, no waiter of this execution can run
+	// again, so the event is buffered for carryover instead.
+	if pendingTasks, ok := ctx.pendingExternalEventTasks[key]; ok && pendingTasks.Len() > 0 &&
+		!ctx.continueAsNewFinalized {
 		task := pendingTasks.Back().Value.(*completableTask)
 		task.complete([]byte(er.Input.GetValue()))
 		return nil
@@ -1697,6 +1707,7 @@ func (ctx *OrchestrationContext) setHistoryLimitFailed(err error) error {
 }
 
 func (ctx *OrchestrationContext) clearCompletionActions() {
+	ctx.continueAsNewFinalized = false
 	for id, action := range ctx.pendingActions {
 		if action.GetCompleteOrchestration() != nil {
 			delete(ctx.pendingActions, id)
@@ -1760,6 +1771,7 @@ func (ctx *OrchestrationContext) setCompleteInternal(
 		completed.Tags = tagcodec.EncodeUserTags(ctx.orchestrationTags)
 	}
 	ctx.pendingActions[sequenceNumber] = completedAction
+	ctx.continueAsNewFinalized = status == protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW
 	return nil
 }
 
@@ -1805,10 +1817,11 @@ func (ctx *OrchestrationContext) actions() []*protos.OrchestratorAction {
 			continue
 		}
 		actions = append(actions, a)
-		if ctx.continuedAsNew && ctx.saveBufferedExternalEvents {
-			if co := a.GetCompleteOrchestration(); co != nil {
-				co.CarryoverEvents = append(co.CarryoverEvents, ctx.unprocessedExternalEvents()...)
-			}
+		// ContinueAsNew intent survives a later failure or termination, so check
+		// the completion's actual status before carrying events forward.
+		if co := a.GetCompleteOrchestration(); co != nil && ctx.saveBufferedExternalEvents &&
+			co.GetOrchestrationStatus() == protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW {
+			co.CarryoverEvents = append(co.CarryoverEvents, ctx.unprocessedExternalEvents()...)
 		}
 	}
 	return actions
diff --git a/task/rewind.go b/task/rewind.go
index 8be8d6d5..e9b0250b 100644
--- a/task/rewind.go
+++ b/task/rewind.go
@@ -36,8 +36,8 @@ func buildRewindResult(id api.InstanceID, oldEvents, newEvents []*protos.History
 		// broader recovery needs dependency-aware history selection.
 		if failureSeen && event.GetOrchestratorStarted() == nil &&
 			event.GetOrchestratorCompleted() == nil && event.GetExecutionCompleted() == nil {
-			return nil, fmt.Errorf("%w: rewind cannot remove an operation failure followed by other durable events (event %d)",
-				api.ErrFeatureNotSupported, event.GetEventId())
+			return nil, fmt.Errorf("Go SDK cannot safely rewind this history: an operation failure is followed by other durable events (event %d): %w",
+				event.GetEventId(), api.ErrFeatureNotSupported)
 		}
 		if failed := event.GetTaskFailed(); failed != nil {
 			failedTasks[failed.TaskScheduledId] = struct{}{}
diff --git a/task/rewind_test.go b/task/rewind_test.go
index 646500ee..98ac6d7b 100644
--- a/task/rewind_test.go
+++ b/task/rewind_test.go
@@ -256,6 +256,8 @@ func TestRewindRejectsHistoryAfterAFailure(t *testing.T) {
 			result, err := NewTaskExecutor(NewTaskRegistry()).ExecuteOrchestrator(context.Background(), "instance",
 				history, []*protos.HistoryEvent{helpers.NewOrchestratorStartedEvent(), rewindEvent("")}, nil)
 			require.ErrorIs(t, err, api.ErrFeatureNotSupported)
+			require.ErrorContains(t, err, "Go SDK cannot safely rewind this history")
+			require.NotContains(t, err.Error(), "not supported by Durable Task Scheduler")
 			require.Nil(t, result)
 			require.True(t, proto.Equal(original, &protos.OrchestratorRequest{PastEvents: history}))
 		})
diff --git a/tests/durabletaskscheduler/dts_continue_as_new_batch_test.go b/tests/durabletaskscheduler/dts_continue_as_new_batch_test.go
new file mode 100644
index 00000000..30c4b487
--- /dev/null
+++ b/tests/durabletaskscheduler/dts_continue_as_new_batch_test.go
@@ -0,0 +1,556 @@
+package durabletaskscheduler_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/microsoft/durabletask-go/api"
+	durabletaskclient "github.com/microsoft/durabletask-go/client"
+	"github.com/microsoft/durabletask-go/durabletaskscheduler"
+	"github.com/microsoft/durabletask-go/internal/protos"
+	"github.com/microsoft/durabletask-go/task"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+// This experiment deliberately fails admission if DTS redelivers the timer
+// separately instead of combining it with queued external events.
+func TestDTSContinueAsNewSameNewEventsBatch(t *testing.T) {
+	if os.Getenv("DTS_CAN_BATCH_TEST") != "1" {
+		t.Skip("set DTS_CAN_BATCH_TEST=1 to run the service batching experiment")
+	}
+	require.True(t, os.Getenv("DTS_CONNECTION_STRING") != "" || os.Getenv("DTS_EMULATOR_ENDPOINT") != "",
+		"the batching experiment requires an explicit DTS target")
+	directory := os.Getenv(continueAsNewArtifactDirEnv)
+	require.NotEmpty(t, directory, "the batching experiment requires an evidence directory")
+	info, err := os.Stat(directory)
+	require.NoError(t, err)
+	require.True(t, info.IsDir())
+
+	options := emulatorOptions(t)
+	runID := uuid.NewString()
+	instanceID := api.InstanceID("go-continue-as-new-batch-" + runID)
+	name := "DTSContinueAsNewBatch-" + runID[:8]
+	registry := task.NewTaskRegistry()
+	require.NoError(t, registry.AddOrchestratorN(name, continueAsNewEventsOrchestrator))
+	events := make([]raisedEvent, 0, 6)
+	expected := make([]string, 0, 6)
+	for index, eventName := range []string{"work", "other", "work", "other", "work", "work"} {
+		event := raisedEvent{name: eventName, payload: fmt.Sprintf("%s-%d", runID, index+1)}
+		events = append(events, event)
+		expected = append(expected, event.name+"="+event.payload)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	logger := api.DefaultLogger()
+	client, err := durabletaskscheduler.NewClient(ctx, options, logger)
+	require.NoError(t, err)
+	recorder := newWorkItemRecorder(string(instanceID))
+	// This profile holds the successful RPC's return, not its invocation.
+	recorder.releaseFirstCompletion()
+	control := newContinueAsNewBatchControl(string(instanceID))
+	outcome := "setup_failed"
+	var workers []*durabletaskclient.TaskHubGrpcWorker
+	t.Cleanup(func() {
+		control.releaseCompletion()
+		cleanupCtx, cancelCleanup := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancelCleanup()
+		cleanupContinueAsNewInstance(t, cleanupCtx, client, instanceID)
+		for _, worker := range workers {
+			shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), 10*time.Second)
+			shutdownErr := worker.Shutdown(shutdownCtx)
+			cancelShutdown()
+			if shutdownErr != nil {
+				t.Errorf("failed to stop experiment worker: %v", shutdownErr)
+			}
+		}
+		if closeErr := client.Close(); closeErr != nil {
+			t.Errorf("failed to close experiment client: %v", closeErr)
+		}
+		recorder.writeEvidence(t)
+		control.writeEvidence(t, outcome)
+	})
+	startWorker := func(label string, gate bool) *durabletaskclient.TaskHubGrpcWorker {
+		workerOptions := *options
+		workerOptions.UnaryInterceptors = append(slices.Clone(options.UnaryInterceptors), recorder.unaryInterceptor())
+		if gate {
+			workerOptions.UnaryInterceptors = append(workerOptions.UnaryInterceptors, control.interceptor())
+		}
+		workerOptions.StreamInterceptors = append(slices.Clone(options.StreamInterceptors), recorder.streamInterceptor(label))
+		worker, workerErr := durabletaskscheduler.NewWorker(&workerOptions, registry, logger,
+			durabletaskclient.WithAutoWorkItemFilters(),
+			durabletaskclient.WithMaxConcurrentOrchestrationWorkItems(1),
+			durabletaskclient.WithWorkerRPCTimeout(time.Minute),
+			durabletaskclient.WithWorkerSilentDisconnectTimeout(time.Minute),
+		)
+		require.NoError(t, workerErr)
+		// Cleanup still needs intake to process termination after a profile timeout.
+		require.NoError(t, worker.Start(context.Background()))
+		workers = append(workers, worker)
+		control.record(label+"_started", "")
+		return worker
+	}
+
+	workerA := startWorker("worker-a", true)
+	_, err = client.ScheduleNewOrchestration(ctx, name, api.WithInstanceID(instanceID),
+		api.WithInput(continueAsNewEventsInput{Expected: len(events)}))
+	require.NoError(t, err)
+	stepCtx, cancelStep := context.WithTimeout(ctx, 30*time.Second)
+	defer cancelStep()
+	var firstResponse *protos.OrchestratorResponse
+	select {
+	case firstResponse = <-control.acknowledged:
+	case <-stepCtx.Done():
+		t.Fatalf("INCONCLUSIVE: initial completion was not acknowledged: %v", stepCtx.Err())
+	}
+	require.Len(t, firstResponse.Actions, 1)
+	require.NotNil(t, firstResponse.Actions[0].GetCreateTimer())
+	timerID := firstResponse.Actions[0].GetId()
+	initialItem, err := recorder.waitForWorkerItem(stepCtx, "worker-a", func(item *observedWorkItem) bool {
+		return executionIDOf(item.history()) != ""
+	})
+	require.NoError(t, err)
+	executionID := executionIDOf(initialItem.history())
+	require.NotEmpty(t, executionID)
+
+	timerItem, err := recorder.waitForWorkerItem(stepCtx, "worker-a", func(item *observedWorkItem) bool {
+		return slices.ContainsFunc(item.Request.NewEvents, isTimerFired)
+	})
+	require.NoError(t, err, "INCONCLUSIVE: no unexecuted timer work item was observed")
+	require.False(t, timerItem.Done)
+	require.NotEmpty(t, timerItem.token)
+	require.Equal(t, executionID, timerItem.Request.GetExecutionId().GetValue())
+	for _, event := range timerItem.Request.NewEvents {
+		if timer := event.GetTimerFired(); timer != nil {
+			require.Equal(t, timerID, timer.GetTimerId())
+			require.True(t, proto.Equal(firstResponse.Actions[0].GetCreateTimer().GetFireAt(), timer.GetFireAt()))
+		}
+	}
+	control.expectAbandonment(timerItem.token)
+	control.record("timer_observed_before_execution", executionID)
+
+	for _, event := range events {
+		raiseErr := client.RaiseEvent(stepCtx, instanceID, event.name, api.WithEventPayload(event.payload))
+		require.NoError(t, raiseErr, "INCONCLUSIVE: a signal was not unambiguously accepted")
+		control.record("event_acknowledged", event.name+"="+event.payload)
+	}
+	// B opens its work-item stream before A releases the outstanding lease.
+	startWorker("worker-b", false)
+	shutdown := make(chan error, 1)
+	control.record("worker-a_shutdown_requested", "")
+	go func() {
+		shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancelShutdown()
+		shutdown <- workerA.Shutdown(shutdownCtx)
+	}()
+	select {
+	case <-control.abandoned:
+	case <-stepCtx.Done():
+		t.Fatalf("INCONCLUSIVE: the timer work item was not successfully abandoned: %v", stepCtx.Err())
+	}
+	require.NoError(t, control.gateFailure(), "INCONCLUSIVE: the completion-return barrier expired")
+	control.releaseCompletion()
+	select {
+	case shutdownErr := <-shutdown:
+		require.NoError(t, shutdownErr)
+	case <-stepCtx.Done():
+		t.Fatalf("INCONCLUSIVE: worker A did not finish draining: %v", stepCtx.Err())
+	}
+
+	redelivered, err := recorder.waitForWorkerItem(stepCtx, "worker-b", func(*observedWorkItem) bool { return true })
+	require.NoError(t, err, "INCONCLUSIVE: worker B received no work item")
+	if err := validateContinueAsNewBatch(redelivered.Request, executionID, timerID, events); err != nil {
+		outcome = "batch_not_exercised"
+		control.record("batch_not_exercised", err.Error())
+		t.Fatalf("BATCH_NOT_EXERCISED: %v", err)
+	}
+	control.record("same_new_events_batch_observed", executionID)
+	outcome = "completion_failed"
+	metadata, err := client.WaitForOrchestrationCompletion(ctx, instanceID, api.WithFetchPayloads(true))
+	require.NoError(t, err)
+	require.Equal(t, api.RUNTIME_STATUS_COMPLETED, metadata.RuntimeStatus)
+	admitted := recorder.requireRegressionCondition(t, events[0].payload)
+	require.Equal(t, "worker-b", admitted.Worker)
+	require.Equal(t, redelivered.token, admitted.token, "the admitted request must be the one that continued as new")
+	require.Equal(t, executionID, executionIDOf(admitted.history()))
+	if admitted.Request.GetRequiresHistoryStreaming() {
+		require.NoError(t, validateUnprocessedBatchHistory(admitted.Streamed))
+	}
+	for _, event := range admitted.history() {
+		require.Nil(t, event.GetExecutionSuspended())
+		require.Nil(t, event.GetExecutionResumed())
+	}
+	require.True(t, slices.ContainsFunc(admitted.history(), func(event *protos.HistoryEvent) bool {
+		return event.GetTimerCreated() != nil && event.GetEventId() == timerID
+	}))
+	finalExecution := recorder.lastExecutionID()
+	require.NotEmpty(t, finalExecution)
+	require.NotEqual(t, executionID, finalExecution)
+	var ledger continueAsNewEventsLedger
+	require.NoError(t, metadata.ReadOutput(&ledger))
+	carryover := carryoverSummaries(t, admitted.Response)
+	outcome = "payload_mismatch"
+	require.Equal(t, expected, ledger.Received,
+		"generation 1 must receive every accepted event exactly once and in order; carryover was %v", carryover)
+	require.False(t, ledger.DeadlineHit)
+	require.Equal(t, expected, carryover)
+	outcome = "passed"
+}
+
+func validateContinueAsNewBatch(
+	request *protos.OrchestratorRequest,
+	executionID string,
+	timerID int32,
+	expected []raisedEvent,
+) error {
+	if request.GetExecutionId().GetValue() != executionID {
+		return errors.New("redelivery changed the execution ID")
+	}
+	if err := validateUnprocessedBatchHistory(request.GetPastEvents()); err != nil {
+		return err
+	}
+	timerSeen := false
+	received := make([]raisedEvent, 0, len(expected))
+	for _, event := range request.GetNewEvents() {
+		switch {
+		case event.GetExecutionSuspended() != nil || event.GetExecutionResumed() != nil:
+			return errors.New("suspension was used")
+		case event.GetTimerFired() != nil:
+			if timerSeen || event.GetTimerFired().GetTimerId() != timerID {
+				return errors.New("unexpected timer in the redelivered batch")
+			}
+			timerSeen = true
+		case event.GetEventRaised() != nil:
+			if !timerSeen {
+				return errors.New("an external event precedes the timer in NewEvents")
+			}
+			var payload string
+			if err := json.Unmarshal([]byte(event.GetEventRaised().GetInput().GetValue()), &payload); err != nil {
+				return fmt.Errorf("invalid event payload: %w", err)
+			}
+			received = append(received, raisedEvent{name: event.GetEventRaised().GetName(), payload: payload})
+		}
+	}
+	if !timerSeen || !slices.Equal(expected, received) {
+		return fmt.Errorf("NewEvents must contain the timer followed by all %d accepted events; observed %s",
+			len(expected), summarizeHistory(request.GetNewEvents()))
+	}
+	return nil
+}
+
+func validateUnprocessedBatchHistory(past []*protos.HistoryEvent) error {
+	for _, event := range past {
+		if event.GetEventRaised() != nil || event.GetTimerFired() != nil ||
+			event.GetExecutionSuspended() != nil || event.GetExecutionResumed() != nil {
+			return errors.New("timer or external events were already committed, or suspension was used")
+		}
+	}
+	return nil
+}
+
+func (r *workItemRecorder) waitForWorkerItem(
+	ctx context.Context,
+	worker string,
+	matches func(*observedWorkItem) bool,
+) (*observedWorkItem, error) {
+	for {
+		r.mu.Lock()
+		for _, item := range r.items {
+			if item.Worker != worker || !matches(item) {
+				continue
+			}
+			snapshot := *item
+			snapshot.Request = proto.Clone(item.Request).(*protos.OrchestratorRequest)
+			snapshot.Streamed = make([]*protos.HistoryEvent, len(item.Streamed))
+			for index, event := range item.Streamed {
+				snapshot.Streamed[index] = proto.Clone(event).(*protos.HistoryEvent)
+			}
+			if item.Response != nil {
+				snapshot.Response = proto.Clone(item.Response).(*protos.OrchestratorResponse)
+			}
+			r.mu.Unlock()
+			return &snapshot, nil
+		}
+		changed := r.changed
+		r.mu.Unlock()
+		select {
+		case <-changed:
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+}
+
+type batchControlObservation struct {
+	Phase  string    `json:"phase"`
+	At     time.Time `json:"at"`
+	Detail string    `json:"detail,omitempty"`
+}
+
+type continueAsNewBatchControl struct {
+	instanceID   string
+	holding      atomic.Bool
+	acknowledged chan *protos.OrchestratorResponse
+	release      chan struct{}
+	releaseOnce  sync.Once
+	abandoned    chan struct{}
+	abandonOnce  sync.Once
+
+	mu           sync.Mutex
+	timerToken   string
+	gateErr      error
+	observations []batchControlObservation
+}
+
+func newContinueAsNewBatchControl(instanceID string) *continueAsNewBatchControl {
+	return &continueAsNewBatchControl{
+		instanceID:   instanceID,
+		acknowledged: make(chan *protos.OrchestratorResponse, 1),
+		release:      make(chan struct{}),
+		abandoned:    make(chan struct{}),
+	}
+}
+
+func (c *continueAsNewBatchControl) releaseCompletion() {
+	c.releaseOnce.Do(func() { close(c.release) })
+}
+
+func (c *continueAsNewBatchControl) expectAbandonment(token string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.timerToken = token
+}
+
+func (c *continueAsNewBatchControl) gateFailure() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.gateErr
+}
+
+func (c *continueAsNewBatchControl) record(phase, detail string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.observations = append(c.observations, batchControlObservation{Phase: phase, At: time.Now().UTC(), Detail: detail})
+}
+
+func (c *continueAsNewBatchControl) interceptor() grpc.UnaryClientInterceptor {
+	return func(
+		ctx context.Context,
+		method string,
+		request any,
+		reply any,
+		connection *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		callOptions ...grpc.CallOption,
+	) error {
+		err := invoker(ctx, method, request, reply, connection, callOptions...)
+		if response, ok := request.(*protos.OrchestratorResponse); ok &&
+			method == protos.TaskHubSidecarService_CompleteOrchestratorTask_FullMethodName &&
+			response.GetInstanceId() == c.instanceID && err == nil && c.holding.CompareAndSwap(false, true) {
+			c.record("initial_completion_acknowledged", "")
+			snapshot := proto.Clone(response).(*protos.OrchestratorResponse)
+			snapshot.CompletionToken = ""
+			c.acknowledged <- snapshot
+			deadline := time.NewTimer(30 * time.Second)
+			defer deadline.Stop()
+			var gateErr error
+			select {
+			case <-c.release:
+			case <-ctx.Done():
+				gateErr = ctx.Err()
+			case <-deadline.C:
+				gateErr = errors.New("the 30-second completion-return barrier expired")
+			}
+			c.mu.Lock()
+			c.gateErr = gateErr
+			c.mu.Unlock()
+			if gateErr != nil {
+				c.record("completion_gate_failed", gateErr.Error())
+			}
+			c.record("initial_completion_return_released", "")
+		}
+		if abandoned, ok := request.(*protos.AbandonOrchestrationTaskRequest); ok &&
+			method == protos.TaskHubSidecarService_AbandonTaskOrchestratorWorkItem_FullMethodName {
+			c.mu.Lock()
+			matches := c.timerToken != "" && abandoned.GetCompletionToken() == c.timerToken
+			c.mu.Unlock()
+			if matches {
+				if err != nil {
+					c.record("timer_abandonment_failed", err.Error())
+				} else {
+					c.record("timer_abandonment_acknowledged", "")
+					c.abandonOnce.Do(func() { close(c.abandoned) })
+				}
+			}
+		}
+		return err
+	}
+}
+
+func (c *continueAsNewBatchControl) writeEvidence(t *testing.T, outcome string) {
+	t.Helper()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	content, err := json.MarshalIndent(struct {
+		Test         string                    `json:"test"`
+		Instance     string                    `json:"instance"`
+		Outcome      string                    `json:"outcome"`
+		Observations []batchControlObservation `json:"observations"`
+	}{t.Name(), c.instanceID, outcome, c.observations}, "", "  ")
+	if err == nil {
+		name := strings.ReplaceAll(t.Name(), "/", "_") + ".control.json"
+		err = os.WriteFile(filepath.Join(os.Getenv(continueAsNewArtifactDirEnv), name), content, 0o600)
+	}
+	if err != nil {
+		t.Errorf("failed to write batch control evidence: %v", err)
+	}
+}
+
+func TestContinueAsNewBatchAdmission(t *testing.T) {
+	const timerID int32 = 7
+	expected := []raisedEvent{{name: "work", payload: "1"}, {name: "other", payload: "2"}}
+	valid := &protos.OrchestratorRequest{
+		ExecutionId: wrapperspb.String("original-execution"),
+		PastEvents: []*protos.HistoryEvent{{
+			EventId:   timerID,
+			EventType: &protos.HistoryEvent_TimerCreated{TimerCreated: &protos.TimerCreatedEvent{}},
+		}},
+		NewEvents: []*protos.HistoryEvent{{
+			EventType: &protos.HistoryEvent_TimerFired{TimerFired: &protos.TimerFiredEvent{TimerId: timerID}},
+		}},
+	}
+	for _, event := range expected {
+		raw, err := json.Marshal(event.payload)
+		require.NoError(t, err)
+		valid.NewEvents = append(valid.NewEvents, &protos.HistoryEvent{
+			EventType: &protos.HistoryEvent_EventRaised{EventRaised: &protos.EventRaisedEvent{
+				Name: event.name, Input: wrapperspb.String(string(raw)),
+			}},
+		})
+	}
+	tests := []struct {
+		name   string
+		mutate func(*protos.OrchestratorRequest)
+	}{
+		{"timer committed in past", func(request *protos.OrchestratorRequest) {
+			request.PastEvents = append(request.PastEvents, request.NewEvents[0])
+			request.NewEvents = request.NewEvents[1:]
+		}},
+		{"timer repeated in past", func(request *protos.OrchestratorRequest) {
+			request.PastEvents = append(request.PastEvents, request.NewEvents[0])
+		}},
+		{"trailing event in past", func(request *protos.OrchestratorRequest) {
+			request.PastEvents = append(request.PastEvents, request.NewEvents[2])
+			request.NewEvents = request.NewEvents[:2]
+		}},
+		{"event before timer", func(request *protos.OrchestratorRequest) {
+			request.NewEvents[0], request.NewEvents[1] = request.NewEvents[1], request.NewEvents[0]
+		}},
+		{"timer alone", func(request *protos.OrchestratorRequest) {
+			request.NewEvents = request.NewEvents[:1]
+		}},
+		{"different execution", func(request *protos.OrchestratorRequest) {
+			request.ExecutionId = wrapperspb.String("new-execution")
+		}},
+		{"different timer", func(request *protos.OrchestratorRequest) {
+			request.NewEvents[0].GetTimerFired().TimerId++
+		}},
+		{"duplicate event", func(request *protos.OrchestratorRequest) {
+			request.NewEvents = append(request.NewEvents, request.NewEvents[1])
+		}},
+		{"reordered events", func(request *protos.OrchestratorRequest) {
+			request.NewEvents[1], request.NewEvents[2] = request.NewEvents[2], request.NewEvents[1]
+		}},
+		{"suspension", func(request *protos.OrchestratorRequest) {
+			request.NewEvents = append(request.NewEvents, &protos.HistoryEvent{
+				EventType: &protos.HistoryEvent_ExecutionSuspended{ExecutionSuspended: &protos.ExecutionSuspendedEvent{}},
+			})
+		}},
+	}
+	require.NoError(t, validateContinueAsNewBatch(valid, "original-execution", timerID, expected))
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			request := proto.Clone(valid).(*protos.OrchestratorRequest)
+			test.mutate(request)
+			require.Error(t, validateContinueAsNewBatch(request, "original-execution", timerID, expected))
+		})
+	}
+}
+
+func TestContinueAsNewBatchControl(t *testing.T) {
+	for _, cancelGate := range []bool{false, true} {
+		t.Run(fmt.Sprintf("cancel=%t", cancelGate), func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			control := newContinueAsNewBatchControl("instance")
+			defer control.releaseCompletion()
+			response := &protos.OrchestratorResponse{InstanceId: "instance", CompletionToken: "initial-token"}
+			invoked := false
+			done := make(chan error, 1)
+			go func() {
+				done <- control.interceptor()(ctx, protos.TaskHubSidecarService_CompleteOrchestratorTask_FullMethodName,
+					response, nil, nil, func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error {
+						invoked = true
+						return nil
+					})
+			}()
+			select {
+			case acknowledged := <-control.acknowledged:
+				require.True(t, invoked)
+				require.Empty(t, acknowledged.CompletionToken)
+				require.Equal(t, "initial-token", response.CompletionToken)
+			case <-ctx.Done():
+				t.Fatal("initial completion acknowledgement was not observed")
+			}
+			select {
+			case <-done:
+				t.Fatal("the initial call returned before the gate was released")
+			default:
+			}
+			if cancelGate {
+				cancel()
+			} else {
+				control.expectAbandonment("timer-token")
+				for _, callErr := range []error{errors.New("transient RPC error"), nil} {
+					err := control.interceptor()(ctx, protos.TaskHubSidecarService_AbandonTaskOrchestratorWorkItem_FullMethodName,
+						&protos.AbandonOrchestrationTaskRequest{CompletionToken: "timer-token"}, nil, nil,
+						func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error { return callErr })
+					require.ErrorIs(t, err, callErr)
+					if callErr != nil {
+						select {
+						case <-control.abandoned:
+							t.Fatal("a failed abandonment was reported as acknowledged")
+						default:
+						}
+					}
+				}
+				select {
+				case <-control.abandoned:
+				case <-ctx.Done():
+					t.Fatal("successful abandonment was not observed")
+				}
+				control.releaseCompletion()
+			}
+			require.NoError(t, <-done, "return the real successful completion result even if the test gate is canceled")
+			if cancelGate {
+				require.ErrorIs(t, control.gateFailure(), context.Canceled)
+			} else {
+				require.NoError(t, control.gateFailure())
+			}
+		})
+	}
+}
diff --git a/tests/durabletaskscheduler/dts_continue_as_new_events_test.go b/tests/durabletaskscheduler/dts_continue_as_new_events_test.go
new file mode 100644
index 00000000..5f02bc65
--- /dev/null
+++ b/tests/durabletaskscheduler/dts_continue_as_new_events_test.go
@@ -0,0 +1,762 @@
+package durabletaskscheduler_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/microsoft/durabletask-go/api"
+	durabletaskclient "github.com/microsoft/durabletask-go/client"
+	"github.com/microsoft/durabletask-go/durabletaskscheduler"
+	"github.com/microsoft/durabletask-go/internal/protos"
+	"github.com/microsoft/durabletask-go/task"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+const (
+	continueAsNewTimer              = time.Second
+	continueAsNewCollectionDeadline = 10 * time.Second
+	continueAsNewArtifactDirEnv     = "DTS_CAN_ARTIFACT_DIR"
+)
+
+type continueAsNewEventsInput struct {
+	Generation int
+	Expected   int
+}
+
+type continueAsNewEventsLedger struct {
+	Received    []string
+	DeadlineHit bool
+}
+
+// continueAsNewEventsOrchestrator leaves a "work" wait pending in generation 0 and
+// continues as new when its timer wins. Generation 1 records the events it
+// receives until it has all of them or its durable deadline fires.
+func continueAsNewEventsOrchestrator(ctx *task.OrchestrationContext) (any, error) {
+	var input continueAsNewEventsInput
+	if err := ctx.GetInput(&input); err != nil {
+		return nil, err
+	}
+	if input.Generation == 0 {
+		pending := ctx.WaitForSingleEvent("work", -1)
+		timer := ctx.CreateTimer(continueAsNewTimer)
+		if ctx.WhenAny(pending, timer) != timer {
+			return nil, errors.New("a work event won before the timer")
+		}
+		ctx.ContinueAsNew(
+			continueAsNewEventsInput{Generation: 1, Expected: input.Expected},
+			task.WithKeepUnprocessedEvents(),
+		)
+		return nil, nil
+	}
+
+	work := task.NewEventChannel[string](ctx, "work")
+	other := task.NewEventChannel[string](ctx, "other")
+	deadline := ctx.CreateTimer(continueAsNewCollectionDeadline)
+	var ledger continueAsNewEventsLedger
+	for len(ledger.Received) < input.Expected && !ledger.DeadlineHit {
+		ctx.Select(
+			task.OnEvent(work, func(value string) { ledger.Received = append(ledger.Received, "work="+value) }),
+			task.OnEvent(other, func(value string) { ledger.Received = append(ledger.Received, "other="+value) }),
+			task.OnTask(deadline, func(task.Task) { ledger.DeadlineHit = true }),
+		)
+	}
+	return ledger, nil
+}
+
+// TestDTSContinueAsNewKeepsEventsRaisedAfterTimerWins is the service-level
+// regression for external events that follow a winning timer at a
+// ContinueAsNew boundary while a WaitForSingleEvent loser is still pending.
+//
+// The service cannot be forced to put the timer and later events in one batch.
+// Instead, the test suspends the instance before the timer exists. The timer
+// and the events raised afterward are then buffered in history, and the resume
+// work item replays them in order after generation 0 finalizes. The test proves
+// that condition from unmodified work items before it checks the outcome.
+func TestDTSContinueAsNewKeepsEventsRaisedAfterTimerWins(t *testing.T) {
+	for _, restartWorker := range []bool{false, true} {
+		name := "same-worker"
+		if restartWorker {
+			name = "fresh-worker-replay"
+		}
+		t.Run(name, func(t *testing.T) {
+			runContinueAsNewEventsScenario(t, restartWorker)
+		})
+	}
+}
+
+func runContinueAsNewEventsScenario(t *testing.T, restartWorker bool) {
+	options := emulatorOptions(t)
+	runID := uuid.NewString()
+	instanceID := api.InstanceID("go-continue-as-new-events-" + runID)
+	orchestratorName := "DTSContinueAsNewEvents-" + runID[:8]
+	registry := task.NewTaskRegistry()
+	require.NoError(t, registry.AddOrchestratorN(orchestratorName, continueAsNewEventsOrchestrator))
+
+	events := make([]raisedEvent, 0, 6)
+	for index, name := range []string{"work", "other", "work", "other", "work", "work"} {
+		events = append(events, raisedEvent{name: name, payload: fmt.Sprintf("%s-%d", runID, index+1)})
+	}
+	expected := make([]string, 0, len(events))
+	for _, event := range events {
+		expected = append(expected, event.name+"="+event.payload)
+	}
+
+	recorder := newWorkItemRecorder(string(instanceID))
+	logger := api.DefaultLogger()
+	client, err := durabletaskscheduler.NewClient(context.Background(), options, logger)
+	require.NoError(t, err)
+
+	var workers []*durabletaskclient.TaskHubGrpcWorker
+	startWorker := func(label string) *durabletaskclient.TaskHubGrpcWorker {
+		workerOptions := *options
+		workerOptions.UnaryInterceptors = append(
+			slices.Clone(options.UnaryInterceptors),
+			recorder.unaryInterceptor(),
+		)
+		workerOptions.StreamInterceptors = append(
+			slices.Clone(options.StreamInterceptors),
+			recorder.streamInterceptor(label),
+		)
+		worker, workerErr := durabletaskscheduler.NewWorker(
+			&workerOptions,
+			registry,
+			logger,
+			durabletaskclient.WithAutoWorkItemFilters(),
+		)
+		require.NoError(t, workerErr)
+		require.NoError(t, worker.Start(context.Background()))
+		workers = append(workers, worker)
+		return worker
+	}
+	stopWorker := func(worker *durabletaskclient.TaskHubGrpcWorker) error {
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		return worker.Shutdown(shutdownCtx)
+	}
+	t.Cleanup(func() {
+		// A held completion keeps a worker drain waiting, so release it first.
+		recorder.releaseFirstCompletion()
+		cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		cleanupContinueAsNewInstance(t, cleanupCtx, client, instanceID)
+		for _, worker := range workers {
+			if err := stopWorker(worker); err != nil {
+				t.Errorf("failed to stop worker: %v", err)
+			}
+		}
+		if err := client.Close(); err != nil {
+			t.Errorf("failed to close client: %v", err)
+		}
+		recorder.writeEvidence(t)
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	worker := startWorker("worker-a")
+	_, err = client.ScheduleNewOrchestration(
+		ctx,
+		orchestratorName,
+		api.WithInstanceID(instanceID),
+		api.WithInput(continueAsNewEventsInput{Expected: len(events)}),
+	)
+	require.NoError(t, err)
+
+	// Suspending while the first completion is held orders the suspension
+	// before the timer, which only exists once that completion commits.
+	stepCtx, cancelStep := context.WithTimeout(ctx, 30*time.Second)
+	defer cancelStep()
+	require.NoError(t, recorder.waitForHeldCompletion(stepCtx))
+	require.NoError(
+		t,
+		client.SuspendOrchestration(ctx, instanceID, "buffer the timer and trailing events"),
+		"INCONCLUSIVE: the service rejected suspension while the first turn was in flight",
+	)
+	recorder.releaseFirstCompletion()
+	require.NoError(t, recorder.waitForSuspendedTimer(stepCtx))
+
+	for _, event := range events {
+		require.NoError(t, client.RaiseEvent(ctx, instanceID, event.name, api.WithEventPayload(event.payload)))
+	}
+	if restartWorker {
+		// RaiseEvent only acknowledges enqueueing. Drain all six signals into
+		// committed history before closing the stream that can claim them.
+		handoffCtx, cancelHandoff := context.WithTimeout(ctx, 30*time.Second)
+		handoffErr := recorder.waitForCommittedEvents(handoffCtx, "worker-a", events)
+		cancelHandoff()
+		require.NoError(t, handoffErr, "INCONCLUSIVE: signals were not committed before worker handoff")
+		require.NoError(t, stopWorker(worker))
+		startWorker("worker-b")
+	}
+	require.NoError(t, client.ResumeOrchestration(ctx, instanceID, "replay the buffered history"))
+
+	metadata, err := client.WaitForOrchestrationCompletion(ctx, instanceID, api.WithFetchPayloads(true))
+	require.NoError(t, err)
+	require.Equal(t, api.RUNTIME_STATUS_COMPLETED, metadata.RuntimeStatus)
+
+	admitted := recorder.requireRegressionCondition(t, events[0].payload)
+	carryover := carryoverSummaries(t, admitted.Response)
+	var ledger continueAsNewEventsLedger
+	require.NoError(t, metadata.ReadOutput(&ledger))
+	require.Equal(
+		t,
+		expected,
+		ledger.Received,
+		"generation 1 must receive every accepted event exactly once and in order; continue-as-new carryover was %v",
+		carryover,
+	)
+	require.False(t, ledger.DeadlineHit)
+	require.Equal(t, expected, carryover)
+
+	firstExecution := executionIDOf(admitted.history())
+	finalExecution := recorder.lastExecutionID()
+	require.NotEmpty(t, firstExecution)
+	require.NotEmpty(t, finalExecution)
+	require.NotEqual(t, firstExecution, finalExecution, "ContinueAsNew must start a new execution")
+	if restartWorker {
+		require.Equal(t, "worker-b", admitted.Worker, "a fresh worker must replay generation 0")
+	}
+}
+
+type raisedEvent struct {
+	name    string
+	payload string
+}
+
+// observedWorkItem is an unmodified copy of one orchestration work item and
+// the completion the worker sent for it.
+type observedWorkItem struct {
+	Worker   string
+	Request  *protos.OrchestratorRequest
+	Streamed []*protos.HistoryEvent
+	Response *protos.OrchestratorResponse
+	Err      error
+	Done     bool
+	token    string
+}
+
+func (item *observedWorkItem) history() []*protos.HistoryEvent {
+	past := item.Request.GetPastEvents()
+	if item.Request.GetRequiresHistoryStreaming() {
+		past = item.Streamed
+	}
+	return append(slices.Clone(past), item.Request.GetNewEvents()...)
+}
+
+// workItemRecorder observes one instance's worker traffic through standard
+// gRPC interceptors. It forwards every message unchanged and can hold the
+// instance's first completion until released.
+type workItemRecorder struct {
+	instanceID  string
+	holding     atomic.Bool
+	held        chan struct{}
+	release     chan struct{}
+	releaseOnce sync.Once
+
+	mu      sync.Mutex
+	changed chan struct{}
+	items   []*observedWorkItem
+}
+
+func newWorkItemRecorder(instanceID string) *workItemRecorder {
+	return &workItemRecorder{
+		instanceID: instanceID,
+		held:       make(chan struct{}),
+		release:    make(chan struct{}),
+		changed:    make(chan struct{}),
+	}
+}
+
+func (r *workItemRecorder) releaseFirstCompletion() {
+	r.releaseOnce.Do(func() { close(r.release) })
+}
+
+func (r *workItemRecorder) waitForHeldCompletion(ctx context.Context) error {
+	select {
+	case <-r.held:
+		return nil
+	case <-ctx.Done():
+		return fmt.Errorf("the first completion was not observed: %w", ctx.Err())
+	}
+}
+
+func (r *workItemRecorder) unaryInterceptor() grpc.UnaryClientInterceptor {
+	return func(
+		ctx context.Context,
+		method string,
+		request any,
+		reply any,
+		connection *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		callOptions ...grpc.CallOption,
+	) error {
+		response, ok := request.(*protos.OrchestratorResponse)
+		if method != protos.TaskHubSidecarService_CompleteOrchestratorTask_FullMethodName ||
+			!ok || response.GetInstanceId() != r.instanceID {
+			return invoker(ctx, method, request, reply, connection, callOptions...)
+		}
+		if r.holding.CompareAndSwap(false, true) {
+			close(r.held)
+			select {
+			case <-r.release:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+		err := invoker(ctx, method, request, reply, connection, callOptions...)
+		r.recordCompletion(proto.Clone(response).(*protos.OrchestratorResponse), err)
+		return err
+	}
+}
+
+func (r *workItemRecorder) streamInterceptor(worker string) grpc.StreamClientInterceptor {
+	return func(
+		ctx context.Context,
+		description *grpc.StreamDesc,
+		connection *grpc.ClientConn,
+		method string,
+		streamer grpc.Streamer,
+		callOptions ...grpc.CallOption,
+	) (grpc.ClientStream, error) {
+		stream, err := streamer(ctx, description, connection, method, callOptions...)
+		if err != nil {
+			return stream, err
+		}
+		if method == protos.TaskHubSidecarService_GetWorkItems_FullMethodName ||
+			method == protos.TaskHubSidecarService_StreamInstanceHistory_FullMethodName {
+			return &observedStream{ClientStream: stream, recorder: r, worker: worker}, nil
+		}
+		return stream, nil
+	}
+}
+
+type observedStream struct {
+	grpc.ClientStream
+	recorder   *workItemRecorder
+	worker     string
+	historyFor string
+}
+
+func (s *observedStream) SendMsg(message any) error {
+	if request, ok := message.(*protos.StreamInstanceHistoryRequest); ok {
+		s.historyFor = request.GetInstanceId()
+	}
+	return s.ClientStream.SendMsg(message)
+}
+
+func (s *observedStream) RecvMsg(message any) error {
+	if err := s.ClientStream.RecvMsg(message); err != nil {
+		return err
+	}
+	switch received := message.(type) {
+	case *protos.WorkItem:
+		s.recorder.recordWorkItem(s.worker, received)
+	case *protos.HistoryChunk:
+		if s.historyFor == s.recorder.instanceID {
+			s.recorder.recordHistoryChunk(received)
+		}
+	}
+	return nil
+}
+
+func (r *workItemRecorder) recordWorkItem(worker string, item *protos.WorkItem) {
+	request := item.GetOrchestratorRequest()
+	if request.GetInstanceId() != r.instanceID {
+		return
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.items = append(r.items, &observedWorkItem{
+		Worker:  worker,
+		Request: proto.Clone(request).(*protos.OrchestratorRequest),
+		token:   item.GetCompletionToken(),
+	})
+	r.notifyLocked()
+}
+
+func (r *workItemRecorder) recordHistoryChunk(chunk *protos.HistoryChunk) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for index := len(r.items) - 1; index >= 0; index-- {
+		item := r.items[index]
+		if !item.Done && item.Request.GetRequiresHistoryStreaming() {
+			for _, event := range chunk.GetEvents() {
+				item.Streamed = append(item.Streamed, proto.Clone(event).(*protos.HistoryEvent))
+			}
+			return
+		}
+	}
+}
+
+func (r *workItemRecorder) recordCompletion(response *protos.OrchestratorResponse, err error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, item := range r.items {
+		// The worker retries transient completion failures with the same token,
+		// so a later attempt replaces a failed one.
+		if item.token == response.GetCompletionToken() && (!item.Done || item.Err != nil) {
+			response.CompletionToken = ""
+			item.Response, item.Err, item.Done = response, err, true
+			r.notifyLocked()
+			return
+		}
+	}
+}
+
+func (r *workItemRecorder) notifyLocked() {
+	close(r.changed)
+	r.changed = make(chan struct{})
+}
+
+// waitForSuspendedTimer waits until the timer is delivered and buffered while
+// the instance is suspended, so generation 0 has not finalized yet.
+func (r *workItemRecorder) waitForSuspendedTimer(ctx context.Context) error {
+	for {
+		r.mu.Lock()
+		for _, item := range r.items {
+			if !item.Done || item.Err != nil || !slices.ContainsFunc(item.Request.GetNewEvents(), isTimerFired) {
+				continue
+			}
+			r.mu.Unlock()
+			if completionOf(item.Response) != nil {
+				return errors.New("INCONCLUSIVE: the timer was processed before the suspension took effect")
+			}
+			return nil
+		}
+		changed := r.changed
+		r.mu.Unlock()
+		select {
+		case <-changed:
+		case <-ctx.Done():
+			return fmt.Errorf("INCONCLUSIVE: the service did not deliver the timer while the instance was suspended: %w", ctx.Err())
+		}
+	}
+}
+
+func (r *workItemRecorder) waitForCommittedEvents(ctx context.Context, worker string, events []raisedEvent) error {
+	expected := make([]raisedEvent, len(events))
+	for index, event := range events {
+		raw, err := json.Marshal(event.payload)
+		if err != nil {
+			return fmt.Errorf("failed to serialize expected event: %w", err)
+		}
+		expected[index] = raisedEvent{name: event.name, payload: string(raw)}
+	}
+	_, err := r.waitForWorkerItem(ctx, worker, func(item *observedWorkItem) bool {
+		if !item.Done || item.Err != nil || item.Response.GetNumEventsProcessed() != nil {
+			return false
+		}
+		history := item.history()
+		for _, event := range expected {
+			if !slices.ContainsFunc(history, func(record *protos.HistoryEvent) bool {
+				raised := record.GetEventRaised()
+				return raised.GetName() == event.name && raised.GetInput().GetValue() == event.payload
+			}) {
+				return false
+			}
+		}
+		return true
+	})
+	return err
+}
+
+// requireRegressionCondition returns the work item that continued generation 0
+// as new after proving, from its unmodified history, that it replayed the timer
+// before a trailing work event while the work wait was still pending.
+func (r *workItemRecorder) requireRegressionCondition(t *testing.T, firstWorkPayload string) *observedWorkItem {
+	t.Helper()
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var admitted *observedWorkItem
+	for _, item := range r.items {
+		if !item.Done || item.Err != nil ||
+			completionOf(item.Response).GetOrchestrationStatus() !=
+				protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW {
+			continue
+		}
+		if admitted != nil {
+			t.Fatalf("INCONCLUSIVE: more than one acknowledged continue-as-new completion was observed")
+		}
+		admitted = item
+	}
+	if admitted == nil {
+		t.Fatalf("INCONCLUSIVE: no acknowledged continue-as-new completion was observed for %s", r.instanceID)
+	}
+
+	history := admitted.history()
+	timerFired := slices.IndexFunc(history, isTimerFired)
+	firstWork := slices.IndexFunc(history, func(event *protos.HistoryEvent) bool {
+		return strings.EqualFold(event.GetEventRaised().GetName(), "work")
+	})
+	if timerFired < 0 || firstWork < timerFired {
+		t.Fatalf(
+			"INCONCLUSIVE: the continue-as-new work item did not replay the timer before a pending work event: %s",
+			summarizeHistory(history),
+		)
+	}
+	if got := raisedPayload(t, history[firstWork]); got != firstWorkPayload {
+		t.Fatalf("INCONCLUSIVE: the first trailing work event was %q, want %q", got, firstWorkPayload)
+	}
+	return admitted
+}
+
+func (r *workItemRecorder) lastExecutionID() string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for index := len(r.items) - 1; index >= 0; index-- {
+		if id := executionIDOf(r.items[index].history()); id != "" {
+			return id
+		}
+	}
+	return ""
+}
+
+// writeEvidence logs the observed timeline when the test fails and writes the
+// unmodified observations when DTS_CAN_ARTIFACT_DIR is set. Completion tokens
+// are never recorded.
+func (r *workItemRecorder) writeEvidence(t *testing.T) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if t.Failed() {
+		for index, item := range r.items {
+			t.Logf("work item %d (%s): %s -> %s", index, item.Worker, summarizeHistory(item.history()), summarizeCompletion(item))
+		}
+	}
+	directory := os.Getenv(continueAsNewArtifactDirEnv)
+	if directory == "" {
+		return
+	}
+	type evidence struct {
+		Worker   string            `json:"worker"`
+		Request  json.RawMessage   `json:"request"`
+		Streamed []json.RawMessage `json:"streamedHistory,omitempty"`
+		Response json.RawMessage   `json:"response,omitempty"`
+		Error    string            `json:"error,omitempty"`
+	}
+	records := make([]evidence, 0, len(r.items))
+	for _, item := range r.items {
+		record := evidence{Worker: item.Worker, Request: protoJSON(item.Request)}
+		for _, event := range item.Streamed {
+			record.Streamed = append(record.Streamed, protoJSON(event))
+		}
+		if item.Response != nil {
+			record.Response = protoJSON(item.Response)
+		}
+		if item.Err != nil {
+			record.Error = item.Err.Error()
+		}
+		records = append(records, record)
+	}
+	content, err := json.MarshalIndent(map[string]any{
+		"test":      t.Name(),
+		"failed":    t.Failed(),
+		"instance":  r.instanceID,
+		"workItems": records,
+	}, "", "  ")
+	if err == nil {
+		name := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name()) + ".json"
+		err = os.WriteFile(filepath.Join(directory, name), content, 0o600)
+	}
+	if err != nil {
+		t.Errorf("failed to write continue-as-new evidence: %v", err)
+	}
+}
+
+func cleanupContinueAsNewInstance(t *testing.T, ctx context.Context, client *durabletaskscheduler.Client, id api.InstanceID) {
+	t.Helper()
+	metadata, err := client.FetchOrchestrationMetadata(ctx, id)
+	if errors.Is(err, api.ErrInstanceNotFound) {
+		return
+	}
+	if err != nil {
+		t.Errorf("failed to read %s before cleanup: %v", id, err)
+		return
+	}
+	terminal := slices.Contains([]api.OrchestrationStatus{
+		api.RUNTIME_STATUS_COMPLETED,
+		api.RUNTIME_STATUS_FAILED,
+		api.RUNTIME_STATUS_TERMINATED,
+	}, metadata.RuntimeStatus)
+	if !terminal {
+		if err := client.TerminateOrchestration(ctx, id, api.WithRecursiveTerminate(false)); err != nil {
+			t.Errorf("failed to terminate %s: %v", id, err)
+			return
+		}
+		if _, err := client.WaitForOrchestrationCompletion(ctx, id); err != nil {
+			t.Errorf("failed to wait for %s to terminate: %v", id, err)
+			return
+		}
+	}
+	if _, err := client.PurgeInstances(ctx, api.PurgeInstancesRequest{InstanceIDs: []api.InstanceID{id}}); err != nil {
+		t.Errorf("failed to purge %s: %v", id, err)
+	}
+}
+
+func completionOf(response *protos.OrchestratorResponse) *protos.CompleteOrchestrationAction {
+	for _, action := range response.GetActions() {
+		if completion := action.GetCompleteOrchestration(); completion != nil {
+			return completion
+		}
+	}
+	return nil
+}
+
+func carryoverSummaries(t *testing.T, response *protos.OrchestratorResponse) []string {
+	t.Helper()
+	var carryover []string
+	for _, event := range completionOf(response).GetCarryoverEvents() {
+		carryover = append(carryover, event.GetEventRaised().GetName()+"="+raisedPayload(t, event))
+	}
+	return carryover
+}
+
+func raisedPayload(t *testing.T, event *protos.HistoryEvent) string {
+	t.Helper()
+	var payload string
+	require.NoError(t, json.Unmarshal([]byte(event.GetEventRaised().GetInput().GetValue()), &payload))
+	return payload
+}
+
+func executionIDOf(history []*protos.HistoryEvent) string {
+	for _, event := range history {
+		if started := event.GetExecutionStarted(); started != nil {
+			return started.GetOrchestrationInstance().GetExecutionId().GetValue()
+		}
+	}
+	return ""
+}
+
+func isTimerFired(event *protos.HistoryEvent) bool {
+	return event.GetTimerFired() != nil
+}
+
+func summarizeHistory(history []*protos.HistoryEvent) string {
+	summaries := make([]string, 0, len(history))
+	for _, event := range history {
+		switch {
+		case event.GetExecutionStarted() != nil:
+			summaries = append(summaries, "ExecutionStarted("+executionIDOf([]*protos.HistoryEvent{event})+")")
+		case event.GetTimerCreated() != nil:
+			summaries = append(summaries, fmt.Sprintf("TimerCreated(%d)", event.GetEventId()))
+		case event.GetTimerFired() != nil:
+			summaries = append(summaries, fmt.Sprintf("TimerFired(%d)", event.GetTimerFired().GetTimerId()))
+		case event.GetEventRaised() != nil:
+			raised := event.GetEventRaised()
+			summaries = append(summaries, "EventRaised("+raised.GetName()+"="+raised.GetInput().GetValue()+")")
+		default:
+			kind := fmt.Sprintf("%T", event.GetEventType())
+			summaries = append(summaries, strings.TrimPrefix(kind, "*protos.HistoryEvent_"))
+		}
+	}
+	return strings.Join(summaries, ", ")
+}
+
+func summarizeCompletion(item *observedWorkItem) string {
+	switch {
+	case !item.Done:
+		return "not completed"
+	case item.Err != nil:
+		return "completion failed: " + item.Err.Error()
+	}
+	completion := completionOf(item.Response)
+	if completion == nil {
+		return fmt.Sprintf("%d action(s)", len(item.Response.GetActions()))
+	}
+	return fmt.Sprintf("%s with carryover [%s]", completion.GetOrchestrationStatus(), summarizeHistory(completion.GetCarryoverEvents()))
+}
+
+func protoJSON(message proto.Message) json.RawMessage {
+	content, err := protojson.Marshal(message)
+	if err != nil {
+		return json.RawMessage(fmt.Sprintf("%q", err.Error()))
+	}
+	return content
+}
+
+func TestWorkItemRecorderWaitForCommittedEvents(t *testing.T) {
+	events := []raisedEvent{{name: "work", payload: "one"}, {name: "work", payload: "two"}}
+	history := make([]*protos.HistoryEvent, len(events))
+	for index, event := range events {
+		raw, err := json.Marshal(event.payload)
+		require.NoError(t, err)
+		history[index] = &protos.HistoryEvent{
+			EventType: &protos.HistoryEvent_EventRaised{EventRaised: &protos.EventRaisedEvent{
+				Name: event.name, Input: wrapperspb.String(string(raw)),
+			}},
+		}
+	}
+	tests := []struct {
+		name     string
+		worker   string
+		old      []*protos.HistoryEvent
+		new      []*protos.HistoryEvent
+		streamed bool
+		attempts []error
+		partial  bool
+		ready    bool
+	}{
+		{name: "received but not completed", new: history},
+		{name: "last event still queued", new: history[:1], attempts: []error{nil}},
+		{name: "completion failed", new: history, attempts: []error{errors.New("unavailable")}},
+		{name: "partial completion", new: history, attempts: []error{nil}, partial: true},
+		{name: "different worker", worker: "worker-b", new: history, attempts: []error{nil}},
+		{name: "all events committed", new: history, attempts: []error{nil}, ready: true},
+		{name: "old and new events", old: history[:1], new: history[1:], attempts: []error{nil}, ready: true},
+		{name: "streamed history", old: history[:1], new: history[1:], streamed: true, attempts: []error{nil}, ready: true},
+		{name: "completion retry succeeded", new: history, attempts: []error{errors.New("unavailable"), nil}, ready: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			recorder := newWorkItemRecorder("instance")
+			worker := test.worker
+			if worker == "" {
+				worker = "worker-a"
+			}
+			request := &protos.OrchestratorRequest{
+				InstanceId: "instance", PastEvents: test.old, NewEvents: test.new,
+				RequiresHistoryStreaming: test.streamed,
+			}
+			if test.streamed {
+				request.PastEvents = nil
+			}
+			recorder.recordWorkItem(worker, &protos.WorkItem{
+				CompletionToken: "token",
+				Request:         &protos.WorkItem_OrchestratorRequest{OrchestratorRequest: request},
+			})
+			if test.streamed {
+				recorder.recordHistoryChunk(&protos.HistoryChunk{Events: test.old})
+			}
+			for _, err := range test.attempts {
+				response := &protos.OrchestratorResponse{InstanceId: "instance", CompletionToken: "token"}
+				if test.partial {
+					response.NumEventsProcessed = wrapperspb.Int32(1)
+				}
+				recorder.recordCompletion(response, err)
+			}
+			// A canceled context makes an unsatisfied barrier fail immediately;
+			// already acknowledged history can satisfy it without waiting.
+			ctx, cancel := context.WithCancel(context.Background())
+			cancel()
+			err := recorder.waitForCommittedEvents(ctx, "worker-a", events)
+			if test.ready {
+				require.NoError(t, err)
+			} else {
+				require.ErrorIs(t, err, context.Canceled)
+			}
+		})
+	}
+}
diff --git a/tests/durabletaskscheduler/dts_rewind_test.go b/tests/durabletaskscheduler/dts_rewind_test.go
index 9cefe13c..c8ce02a3 100644
--- a/tests/durabletaskscheduler/dts_rewind_test.go
+++ b/tests/durabletaskscheduler/dts_rewind_test.go
@@ -174,7 +174,7 @@ func TestDTSRewindRejectsHandledFailure(t *testing.T) {
 			err := invoke(ctx, method, request, reply, connection, callOptions...)
 			if response, ok := request.(*protos.OrchestratorResponse); ok && err == nil {
 				for _, action := range response.Actions {
-					if strings.Contains(action.GetCompleteOrchestration().GetFailureDetails().GetErrorMessage(), "rewind cannot remove an operation failure") {
+					if strings.Contains(action.GetCompleteOrchestration().GetFailureDetails().GetErrorMessage(), "Go SDK cannot safely rewind this history") {
 						select {
 						case rejected <- struct{}{}:
 						default: