Add exhaustive natural XDC buffered event coverage (#11789)

## Summary - add transition-history XDC E2E coverage for every buffered event that current production code can emit - create buffered events naturally through workflow/activity APIs, workflow commands, transfer and timer tasks, child callbacks, Nexus callbacks, and conflict reapplication - hold a workflow task, fail over, let its timeout flush the old active cluster's buffer, and then release replication to exercise conflict resolution - prove every flushed event is persisted with a real event ID on the non-current losing history branch - verify cherry-pickable inputs are reapplied to the winning branch and branch-dependent outcomes are skipped - reproduce the Nexus #10986 conflict shape: reapply a shared-operation completion while skipping an operation that exists only on the losing branch - keep all test infrastructure in `buffered_events_replication_helpers_test.go`; no production or testcore code is changed ## Coverage The scenarios naturally cover all 30 production-reachable buffered event types: - activity started, completed, failed, timed out, and canceled - timer fired - workflow cancel requested and signaled - workflow options updated, paused, and unpaused - external signal/cancel success and failure callbacks - child start failure, child started, and all five child terminal outcomes - update admitted through real conflict reapplication - Nexus started, completed, failed, canceled, timed out, and both cancel-request outcomes Four values in the buffered-event set cannot be naturally buffered on current `main` and are intentionally not fabricated: - `WORKFLOW_EXECUTION_UPDATE_REJECTED`, `WORKFLOW_PROPERTIES_MODIFIED_EXTERNALLY`, and `ACTIVITY_PROPERTIES_MODIFIED_EXTERNALLY` have no production emitter - `WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED` has an emitter, but its production precondition rejects workflows with a pending workflow task ## What the tests prove - expected events first exist in mutable-state buffering with `BufferedEventID` - failover creates a winning branch before the old active cluster's replication is released - workflow-task timeout flushes the old active cluster's buffer - every expected event is found on the non-current losing branch with a positive, non-buffered event ID - a naturally buffered marker signal proves the losing batch passed through conflict reapplication - signals and updates that are eligible for reapplication reach the winner - activity, timer, child, external-command-result, Nexus cancel-request-result, and losing-only Nexus events remain on the losing branch when the winner lacks the state needed to apply them - shared Nexus operation outcomes are reapplied by scheduled event ID, while losing-only operations are skipped - current histories converge after the relevant replication tasks are released ## Determinism - replication is intercepted and blocked only for the workflow under test - the next workflow task is explicitly created and polled before callbacks are released - Nexus handlers use response barriers released only after the held workflow task is confirmed - losing branches are read directly from persistence and identified by their expected events - assertions poll observable conditions instead of relying on fixed sleeps - the pause feature flag is overridden only for the test that exercises pause/unpause - scenarios skip when transition history is disabled; Nexus conflict scenarios also skip for the CHASM implementation ## Validation - repository `gci` formatting and `git diff --check` - `GOLANGCI_LINT_FIX=false GOLANGCI_LINT_BASE_REV=HEAD~ make lint-code` - XDC package compilation with `test_dep` - focused transition-history E2E runs for mixed inputs/update reapplication, activity outcomes, child outcomes, external workflow outcomes, and Nexus outcomes - CI PostgreSQL XDC, unit, integration, mixed-brain, formatting, and all linter checks pass
2026-08-30 18:41:49 -07:00 · 2026-08-27 08:07:00 -07:00
parent 38948f8a27
commit 4d0afa1cb7
3 changed files with 2160 additions and 0 deletions
--- a/tests/xdc/buffered_events_replication_helpers_test.go
+++ b/tests/xdc/buffered_events_replication_helpers_test.go
--- a/tests/xdc/buffered_events_replication_test.go
+++ b/tests/xdc/buffered_events_replication_test.go
@@ -0,0 +1,451 @@
+package xdc
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	commandpb "go.temporal.io/api/command/v1"
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	"go.temporal.io/server/common/testing/await"
+)
+
+// Four values in HistoryBuilder's buffered-event set cannot be naturally
+// buffered on current main. EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_REJECTED,
+// EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED_EXTERNALLY, and
+// EVENT_TYPE_ACTIVITY_PROPERTIES_MODIFIED_EXTERNALLY have no production emitter.
+// EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED has an emitter, but
+// its production precondition rejects every workflow with a pending workflow
+// task, so it cannot enter the buffer. The tests below cover every other value
+// through its production API, transfer task, timer, callback, or reapplier.
+
+// TestNaturallyBufferedInputsFlushedAndReappliedAfterFailover buffers an activity result, timer,
+// cancel request, signal, options update, pause, and unpause behind one workflow task; conflict
+// reapplication then buffers an update-admitted event on the winner. It expects external inputs and
+// the update to reach the winner while activity, timer, and pause state remain only on the losing branch.
+func (s *FunctionalClustersTestSuite) TestNaturallyBufferedInputsFlushedAndReappliedAfterFailover() {
+	if !s.enableTransitionHistory {
+		s.T().Skip("buffered event state-based replication requires transition history")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+	s.enableWorkflowPauseForTest()
+
+	// Phase 1: establish identical history with one pending activity.
+	execution, taskQueue := s.startWorkflowWithPendingActivity(ctx, ns)
+	workflowID := execution.WorkflowId
+	s.waitForClusterSynced()
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		require.NotEmpty(t, history)
+		require.Equal(t, enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED, history[len(history)-1].EventType)
+	}, replicationWaitTime, replicationCheckInterval)
+
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+
+	// Phase 2: hold a workflow task and create the losing branch's buffered events.
+	updateID := "buffered-update-" + uuid.NewString()
+	s.acceptUpdateAndStartTimer(ctx, ns, execution, taskQueue, updateID)
+
+	heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.Require().NotEmpty(heldWorkflowTask.TaskToken)
+
+	optionsRequestID := s.completeActivityAndBufferExternalEvents(ctx, ns, execution, taskQueue)
+	naturallyBufferedTypes := []enumspb.EventType{
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
+		enumspb.EVENT_TYPE_TIMER_FIRED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED,
+	}
+	bufferedInputs := bufferedInputsExpectation{
+		Namespace:        namespace,
+		Execution:        execution,
+		UpdateID:         updateID,
+		OptionsRequestID: optionsRequestID,
+		EventTypes:       naturallyBufferedTypes,
+	}
+	s.assertBufferedEventTypes(ctx, bufferedEventExpectation{
+		Namespace:  ns,
+		Execution:  execution,
+		EventTypes: naturallyBufferedTypes,
+	})
+
+	// Phase 3: fail over and create a winning branch on the new active cluster.
+	s.failoverToNewActiveCluster(ctx, ns)
+	s.writeSignalOnNewActive(ctx, activeClusterSignal{
+		Namespace:  namespace,
+		Execution:  execution,
+		SignalName: "winner-signal",
+	})
+	winnerWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 1, ns, taskQueue)
+	s.Require().NotEmpty(winnerWorkflowTask.TaskToken)
+
+	// Phase 4: resolve the conflict and verify losing-branch storage versus reapplication.
+	s.releaseReplicationTask(ctx, replicationToOldActive)
+	s.assertNoBufferedEvents(ctx, 0, ns, execution)
+	s.assertBufferedEventsPersistedOnLosingBranch(ctx, bufferedInputs)
+
+	// Reapplying the losing update while the winner's workflow task is running
+	// naturally creates UpdateAdmitted as a buffered event.
+	for attempt := 0; attempt < 10 && !s.hasBufferedEventType(ctx, 1, ns, execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED); attempt++ {
+		s.releaseReplicationTask(ctx, replicationToNewActive)
+	}
+	s.Require().True(s.hasBufferedEventType(ctx, 1, ns, execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED))
+	await.Require(ctx, s.T(), func(t *await.T) {
+		require.False(t, s.hasBufferedEventType(t.Context(), 1, ns, execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED))
+	}, 45*time.Second, replicationCheckInterval)
+	// The timeout flush and conflict resolution can produce more than one
+	// state-based task. Release only this workflow's tasks until all reapplied
+	// inputs reach the active cluster.
+	for attempt := 0; attempt < 10 && !s.hasReappliedBufferedInputs(ctx, bufferedInputs); attempt++ {
+		s.releaseReplicationTask(ctx, replicationToNewActive)
+	}
+	s.Require().True(s.hasReappliedBufferedInputs(ctx, bufferedInputs))
+	for attempt := 0; attempt < 10 && !s.bufferedEventsHistoriesEqual(ctx, ns, execution); attempt++ {
+		s.releaseReplicationTask(ctx, replicationToOldActive)
+	}
+
+	await.Require(ctx, s.T(), func(t *await.T) {
+		sourceHistory := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, execution)
+		targetHistory := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		require.Equal(t, targetHistory, sourceHistory)
+	}, replicationWaitTime, replicationCheckInterval)
+
+	finalHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	s.Require().Equal(1, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED))
+	s.Require().Equal(1, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED))
+	s.Require().Equal(1, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED))
+	s.Require().Equal(2, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED))
+	assertOnlyExpectedBufferedEventsReapplied(s.T(), finalHistory, naturallyBufferedTypes)
+}
+
+// TestNaturallyBufferedActivityOutcomesFlushedToLosingBranch buffers started, completed, failed,
+// timed-out, and canceled activity outcomes. It expects all outcomes on the losing branch and none
+// to be reapplied to the winner, apart from the winner independently producing its own timeout.
+func (s *FunctionalClustersTestSuite) TestNaturallyBufferedActivityOutcomesFlushedToLosingBranch() {
+	if !s.enableTransitionHistory {
+		s.T().Skip("buffered event state-based replication requires transition history")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+
+	workflowID := "buffered-activities-xdc-" + uuid.NewString()
+	workflowQueue := &taskqueuepb.TaskQueue{Name: workflowID + "-workflow"}
+	execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: workflowID,
+		TaskQueue:  workflowQueue,
+	})
+	firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue)
+
+	activityQueues := map[string]*taskqueuepb.TaskQueue{}
+	var scheduleCommands []*commandpb.Command
+	for _, activityID := range []string{"started", "completed", "failed", "canceled"} {
+		activityQueues[activityID] = &taskqueuepb.TaskQueue{Name: workflowID + "-" + activityID}
+		scheduleCommands = append(scheduleCommands, scheduleActivityCommand(activityID, activityQueues[activityID], time.Minute))
+	}
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     firstTask,
+		Commands: scheduleCommands,
+	})
+
+	// Start the activity that will be canceled before establishing the common
+	// prefix, so its cancellation request can be issued by a workflow command.
+	canceledTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["canceled"])
+	s.signalWorkflow(ctx, workflowSignal{
+		Namespace:  ns,
+		Execution:  execution,
+		SignalName: "prepare-activity-cancellation",
+	})
+	cancelCommandTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue)
+	s.Require().NotNil(cancelCommandTask.History)
+	canceledScheduledID := findActivityScheduledEventID(cancelCommandTask.History.Events, "canceled")
+	s.Require().Positive(canceledScheduledID)
+	s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{
+		Task: cancelCommandTask,
+		Commands: []*commandpb.Command{
+			requestCancelActivityCommand(canceledScheduledID),
+			scheduleActivityCommand("timed-out", &taskqueuepb.TaskQueue{Name: workflowID + "-unpolled"}, 5*time.Second),
+		},
+	})
+
+	s.waitForClusterSynced()
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		require.Positive(t, findActivityScheduledEventID(history, "canceled"))
+	}, replicationWaitTime, replicationCheckInterval)
+
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+	heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue)
+	s.Require().NotEmpty(heldWorkflowTask.TaskToken)
+
+	startedTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["started"])
+	s.Require().NotEmpty(startedTask.TaskToken)
+	completedTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["completed"])
+	s.completeActivityTask(ctx, completedTask)
+	failedTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["failed"])
+	s.failActivityTask(ctx, failedTask)
+	s.cancelActivityTask(ctx, canceledTask)
+
+	expectedTypes := []enumspb.EventType{
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_FAILED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
+	}
+	s.assertBufferedEventTypes(ctx, bufferedEventExpectation{
+		Namespace:  ns,
+		Execution:  execution,
+		EventTypes: expectedTypes,
+	})
+	s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{
+		Namespace:              namespace,
+		Execution:              execution,
+		ReplicationToOldActive: replicationToOldActive,
+		ReplicationToNewActive: replicationToNewActive,
+		ExpectedEventTypes:     expectedTypes,
+		WinnerSignal:           "activity-winner-signal",
+	})
+	winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	for _, eventType := range []enumspb.EventType{
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_FAILED,
+		enumspb.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
+	} {
+		s.Require().Zero(countBufferedEventType(winningHistory, eventType), "%s must remain only on the losing branch", eventType)
+	}
+	s.Require().Equal(
+		1,
+		countBufferedEventType(winningHistory, enumspb.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT),
+		"the winner may produce its own timeout, but must not contain a duplicate reapplied timeout",
+	)
+}
+
+// TestNaturallyBufferedChildWorkflowOutcomesFlushedToLosingBranch buffers child start failure,
+// started, completed, failed, canceled, timed-out, and terminated callbacks for children created only
+// on the losing branch. It expects every callback to be persisted there and skipped on the winner.
+func (s *FunctionalClustersTestSuite) TestNaturallyBufferedChildWorkflowOutcomesFlushedToLosingBranch() {
+	if !s.enableTransitionHistory {
+		s.T().Skip("buffered event state-based replication requires transition history")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+
+	workflowID := "buffered-children-xdc-" + uuid.NewString()
+	parentQueue := &taskqueuepb.TaskQueue{Name: workflowID + "-parent"}
+	execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: workflowID,
+		TaskQueue:  parentQueue,
+	})
+	firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, parentQueue)
+
+	childIDs := map[string]string{}
+	childQueues := map[string]*taskqueuepb.TaskQueue{}
+	for _, outcome := range []string{"completed", "failed", "canceled", "timed-out", "terminated", "duplicate"} {
+		childIDs[outcome] = workflowID + "-" + outcome
+		childQueues[outcome] = &taskqueuepb.TaskQueue{Name: childIDs[outcome] + "-queue"}
+	}
+	duplicateExecution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: childIDs["duplicate"],
+		TaskQueue:  childQueues["duplicate"],
+	})
+	s.Require().NotEmpty(duplicateExecution.RunId)
+
+	s.waitForClusterSynced()
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		require.NotEmpty(t, history)
+	}, replicationWaitTime, replicationCheckInterval)
+
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+	var commands []*commandpb.Command
+	for _, outcome := range []string{"completed", "failed", "canceled", "timed-out", "terminated", "duplicate"} {
+		runTimeout := time.Minute
+		if outcome == "timed-out" {
+			runTimeout = 5 * time.Second
+		}
+		commands = append(commands, startChildWorkflowCommand(childIDs[outcome], childQueues[outcome], runTimeout))
+	}
+	s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{
+		Task:     firstTask,
+		Commands: commands,
+	})
+	heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, parentQueue)
+	s.Require().NotEmpty(heldWorkflowTask.TaskToken)
+
+	completedTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, childQueues["completed"])
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     completedTask,
+		Commands: []*commandpb.Command{completeWorkflowCommand()},
+	})
+
+	failedTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, childQueues["failed"])
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     failedTask,
+		Commands: []*commandpb.Command{failWorkflowCommand("expected child failure")},
+	})
+
+	s.requestWorkflowCancellationEventually(ctx, workflowCancellation{
+		Namespace: ns,
+		Execution: &commonpb.WorkflowExecution{WorkflowId: childIDs["canceled"]},
+	})
+	canceledTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, childQueues["canceled"])
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     canceledTask,
+		Commands: []*commandpb.Command{cancelWorkflowCommand()},
+	})
+
+	s.terminateWorkflowEventually(ctx, workflowTermination{
+		Namespace: ns,
+		Execution: &commonpb.WorkflowExecution{WorkflowId: childIDs["terminated"]},
+		Reason:    "expected child termination",
+	})
+
+	expectedTypes := []enumspb.EventType{
+		enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_FAILED,
+		enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED,
+		enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED,
+		enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_FAILED,
+		enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_CANCELED,
+		enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TIMED_OUT,
+		enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TERMINATED,
+	}
+	s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{
+		Namespace:  ns,
+		Execution:  execution,
+		EventTypes: expectedTypes,
+	})
+	s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{
+		Namespace:              namespace,
+		Execution:              execution,
+		ReplicationToOldActive: replicationToOldActive,
+		ReplicationToNewActive: replicationToNewActive,
+		ExpectedEventTypes:     expectedTypes,
+		WinnerSignal:           "child-winner-signal",
+	})
+	winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	for _, eventType := range expectedTypes {
+		s.Require().Zero(countBufferedEventType(winningHistory, eventType), "%s has no child initiation on the winning branch and must be skipped", eventType)
+	}
+}
+
+// TestNaturallyBufferedExternalWorkflowOutcomesFlushedToLosingBranch buffers successful and failed
+// signal-external and cancel-external results. It expects every result on the losing branch and none
+// on the winner because the corresponding initiated commands do not exist there.
+func (s *FunctionalClustersTestSuite) TestNaturallyBufferedExternalWorkflowOutcomesFlushedToLosingBranch() {
+	if !s.enableTransitionHistory {
+		s.T().Skip("buffered event state-based replication requires transition history")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+
+	workflowID := "buffered-external-xdc-" + uuid.NewString()
+	workflowQueue := &taskqueuepb.TaskQueue{Name: workflowID + "-source"}
+	execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: workflowID,
+		TaskQueue:  workflowQueue,
+	})
+	firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue)
+	signalTargetID := workflowID + "-signal-target"
+	signalTargetQueue := &taskqueuepb.TaskQueue{Name: signalTargetID + "-queue"}
+	signalTargetExecution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: signalTargetID,
+		TaskQueue:  signalTargetQueue,
+	})
+	s.Require().NotEmpty(signalTargetExecution.RunId)
+	targetTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, signalTargetQueue)
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task: targetTask,
+	})
+	cancelTargetID := workflowID + "-completed-cancel-target"
+	cancelTargetExecution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: cancelTargetID,
+		TaskQueue:  &taskqueuepb.TaskQueue{Name: cancelTargetID + "-queue"},
+	})
+	s.terminateWorkflow(ctx, workflowTermination{
+		Namespace: ns,
+		Execution: cancelTargetExecution,
+		Reason:    "completed target makes external cancellation deterministic",
+	})
+
+	s.waitForClusterSynced()
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		require.NotEmpty(t, history)
+	}, replicationWaitTime, replicationCheckInterval)
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+
+	missingWorkflowID := workflowID + "-missing"
+	missingRunID := uuid.NewString()
+	heldWorkflowTask := s.completeWorkflowTaskAndReturnNext(ctx, workflowTaskCompletion{
+		Task: firstTask,
+		Commands: []*commandpb.Command{
+			signalExternalWorkflowCommand(signalTargetID, "", "successful-external-signal"),
+			signalExternalWorkflowCommand(missingWorkflowID, missingRunID, "failed-external-signal"),
+			cancelExternalWorkflowCommand(missingWorkflowID, missingRunID),
+			cancelExternalWorkflowCommand(cancelTargetID, cancelTargetExecution.RunId),
+		},
+	})
+	s.Require().NotEmpty(heldWorkflowTask.TaskToken)
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, signalTargetExecution)
+		require.True(t, hasSignalNamed(history, "successful-external-signal"))
+	}, replicationWaitTime, replicationCheckInterval)
+
+	expectedTypes := []enumspb.EventType{
+		enumspb.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED,
+		enumspb.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED,
+		enumspb.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_SIGNALED,
+		enumspb.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_CANCEL_REQUESTED,
+	}
+	s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{
+		Namespace:  ns,
+		Execution:  execution,
+		EventTypes: expectedTypes,
+	})
+	s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{
+		Namespace:              namespace,
+		Execution:              execution,
+		ReplicationToOldActive: replicationToOldActive,
+		ReplicationToNewActive: replicationToNewActive,
+		ExpectedEventTypes:     expectedTypes,
+		WinnerSignal:           "external-winner-signal",
+	})
+	winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	for _, eventType := range expectedTypes {
+		s.Require().Zero(countBufferedEventType(winningHistory, eventType), "%s must remain only on the losing branch", eventType)
+	}
+}
--- a/tests/xdc/buffered_nexus_events_replication_test.go
+++ b/tests/xdc/buffered_nexus_events_replication_test.go
@@ -0,0 +1,384 @@
+package xdc
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/nexus-rpc/sdk-go/nexus"
+	"github.com/stretchr/testify/require"
+	commandpb "go.temporal.io/api/command/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	commonnexus "go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/common/nexus/nexustest"
+	"go.temporal.io/server/common/testing/await"
+)
+
+// TestBufferedNexusEventsReapplySharedOperationAndSkipLosingOnlyOperation buffers completion of an
+// operation shared by both branches plus start and completion of an operation created only on the
+// losing branch. It expects all three events on the loser, only the shared completion on the winner,
+// and the losing-only operation skipped. This covers temporalio/temporal#10986.
+func (s *NexusStateReplicationSuite) TestBufferedNexusEventsReapplySharedOperationAndSkipLosingOnlyOperation() {
+	if !s.enableTransitionHistory || s.chasmEnabled {
+		s.T().Skip("this conflict-reapplication regression is specific to transition-history HSM Nexus operations")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+
+	// Phase 1: establish identical history with one shared, started Nexus operation.
+	endpointName, operationCallbacks, allowLosingOnlyOperationStart := s.setupBufferedNexusEndpoint(ctx)
+
+	workflowID := "buffered-nexus-conflict-" + uuid.NewString()
+	taskQueue := &taskqueuepb.TaskQueue{Name: "buffered-nexus-conflict", Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
+	execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: workflowID,
+		TaskQueue:  taskQueue,
+	})
+
+	firstWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     firstWorkflowTask,
+		Commands: []*commandpb.Command{scheduleBufferedNexusOperationCommand(endpointName, "shared-operation")},
+	})
+	sharedOperationCallback := receiveBufferedNexusCallback(ctx, s.T(), operationCallbacks, "shared-operation")
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, execution)
+		sharedOperationScheduledEventID := findNexusScheduledEventID(history, "shared-operation")
+		require.Positive(t, sharedOperationScheduledEventID)
+		require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, sharedOperationScheduledEventID))
+	}, replicationWaitTime, replicationCheckInterval)
+	sharedOperationScheduledEventID := findNexusScheduledEventID(s.getWorkflowHistory(ctx, s.T(), 0, ns, execution), "shared-operation")
+	s.Require().Positive(sharedOperationScheduledEventID)
+	s.waitForClusterSynced()
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, sharedOperationScheduledEventID))
+	}, replicationWaitTime, replicationCheckInterval)
+
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+
+	// Phase 2: hold a workflow task and buffer completions for shared and losing-only operations.
+	s.signalWorkflow(ctx, workflowSignal{
+		Namespace:  ns,
+		Execution:  execution,
+		SignalName: "create-losing-operation",
+	})
+	scheduleLosingOnlyOperationTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{
+		Task:     scheduleLosingOnlyOperationTask,
+		Commands: []*commandpb.Command{scheduleBufferedNexusOperationCommand(endpointName, "losing-only-operation")},
+	})
+	heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.Require().NotEmpty(heldWorkflowTask.TaskToken)
+	losingOnlyOperationScheduledEventID := findNexusScheduledEventID(heldWorkflowTask.History.Events, "losing-only-operation")
+	s.Require().Positive(losingOnlyOperationScheduledEventID)
+	close(allowLosingOnlyOperationStart)
+	losingOnlyOperationCallback := receiveBufferedNexusCallback(ctx, s.T(), operationCallbacks, "losing-only-operation")
+
+	s.completeBufferedNexusOperation(ctx, bufferedNexusOperationCompletion{
+		Callback: sharedOperationCallback,
+		Result:   "shared-result",
+	})
+	s.completeBufferedNexusOperation(ctx, bufferedNexusOperationCompletion{
+		Callback: losingOnlyOperationCallback,
+		Result:   "losing-result",
+	})
+	s.assertBufferedEventTypes(ctx, bufferedEventExpectation{
+		Namespace: ns,
+		Execution: execution,
+		EventTypes: []enumspb.EventType{
+			enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED,
+			enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED,
+			enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED,
+		},
+	})
+
+	// Phase 3: fail over and create a winning branch on the new active cluster.
+	s.failoverToNewActiveCluster(ctx, ns)
+	s.writeSignalOnNewActive(ctx, activeClusterSignal{
+		Namespace:  namespace,
+		Execution:  execution,
+		SignalName: "nexus-winner-signal",
+	})
+
+	// Phase 4: resolve the conflict and verify losing-branch storage versus reapplication.
+	s.releaseReplicationTask(ctx, replicationToOldActive)
+	s.assertNoBufferedEvents(ctx, 0, ns, execution)
+	losingHistory := s.findBufferedNexusLosingBranch(ctx, bufferedNexusLosingBranch{
+		Namespace:        namespace,
+		Execution:        execution,
+		ScheduledEventID: losingOnlyOperationScheduledEventID,
+	})
+	s.Require().True(hasNexusEventForScheduledID(losingHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, losingOnlyOperationScheduledEventID))
+	s.Require().True(hasNexusEventForScheduledID(losingHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, losingOnlyOperationScheduledEventID))
+	s.Require().Equal(2, countBufferedEventType(losingHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED))
+
+	for range 10 {
+		targetHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+		if hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, sharedOperationScheduledEventID) {
+			break
+		}
+		s.releaseReplicationTask(ctx, replicationToNewActive)
+	}
+	targetHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	s.Require().True(hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, sharedOperationScheduledEventID))
+	s.Require().False(hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, losingOnlyOperationScheduledEventID))
+	s.Require().False(hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, losingOnlyOperationScheduledEventID))
+	await.Require(ctx, s.T(), func(t *await.T) {
+		describeResponse, describeErr := s.clusters[1].FrontendClient().DescribeWorkflowExecution(t.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: execution,
+		})
+		require.NoError(t, describeErr)
+		require.Empty(t, describeResponse.PendingNexusOperations)
+	}, replicationWaitTime, replicationCheckInterval)
+
+	for attempt := 0; attempt < 10 && !s.bufferedEventsHistoriesEqual(ctx, ns, execution); attempt++ {
+		s.releaseReplicationTask(ctx, replicationToOldActive)
+	}
+	await.Require(ctx, s.T(), func(t *await.T) {
+		require.True(t, s.bufferedEventsHistoriesEqual(t.Context(), ns, execution))
+	}, replicationWaitTime, replicationCheckInterval)
+}
+
+// TestNaturallyBufferedNexusOutcomesFlushedAndReapplied buffers failed, canceled, timed-out, and
+// cancel-request-failed outcomes for operations shared by both branches. It expects the terminal operation
+// outcomes to be reapplied to the winner and the non-cherry-pickable cancellation result to remain on the loser.
+func (s *NexusStateReplicationSuite) TestNaturallyBufferedNexusOutcomesFlushedAndReapplied() {
+	if !s.enableTransitionHistory || s.chasmEnabled {
+		s.T().Skip("this conflict-reapplication regression is specific to transition-history HSM Nexus operations")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+
+	callbacks := make(chan bufferedNexusCallback, 4)
+	allowCancellationResponse := make(chan struct{})
+	handler := nexustest.Handler{
+		OnStartOperation: func(_ context.Context, _, operation string, _ *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
+			callbacks <- bufferedNexusCallback{
+				operation: operation,
+				url:       options.CallbackURL,
+				token:     options.CallbackHeader.Get(commonnexus.CallbackTokenHeader),
+			}
+			return &nexus.HandlerStartOperationResultAsync{OperationToken: operation}, nil
+		},
+		OnCancelOperation: func(_ context.Context, _, operation, _ string, _ nexus.CancelOperationOptions) error {
+			<-allowCancellationResponse
+			if operation == "cancel-failed" {
+				return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "expected cancellation failure")
+			}
+			return nil
+		},
+	}
+	endpointName := s.createBufferedNexusEndpoint(ctx, handler)
+
+	workflowID := "buffered-nexus-outcomes-" + uuid.NewString()
+	taskQueue := &taskqueuepb.TaskQueue{Name: workflowID, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
+	execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: workflowID,
+		TaskQueue:  taskQueue,
+	})
+	firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	operations := []string{"failed", "canceled", "timed-out", "cancel-failed"}
+	commands := make([]*commandpb.Command, 0, len(operations))
+	for _, operation := range operations {
+		timeout := time.Minute
+		if operation == "timed-out" {
+			timeout = 5 * time.Second
+		}
+		commands = append(commands, scheduleBufferedNexusOperationCommandWithTimeout(endpointName, operation, timeout))
+	}
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     firstTask,
+		Commands: commands,
+	})
+
+	operationCallbacks := make(map[string]bufferedNexusCallback, len(operations))
+	for range operations {
+		callback := receiveAnyBufferedNexusCallback(ctx, s.T(), callbacks)
+		operationCallbacks[callback.operation] = callback
+	}
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, execution)
+		for _, operation := range operations {
+			scheduledID := findNexusScheduledEventID(history, operation)
+			require.Positive(t, scheduledID)
+			require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, scheduledID))
+		}
+	}, replicationWaitTime, replicationCheckInterval)
+	s.waitForClusterSynced()
+	await.Require(ctx, s.T(), func(t *await.T) {
+		history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution)
+		for _, operation := range operations {
+			scheduledID := findNexusScheduledEventID(history, operation)
+			require.Positive(t, scheduledID)
+			require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, scheduledID))
+		}
+	}, replicationWaitTime, replicationCheckInterval)
+	await.Require(ctx, s.T(), func(t *await.T) {
+		response, describeErr := s.clusters[1].FrontendClient().DescribeWorkflowExecution(t.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: execution,
+		})
+		require.NoError(t, describeErr)
+		require.Len(t, response.PendingNexusOperations, len(operations))
+	}, replicationWaitTime, replicationCheckInterval)
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+
+	triggerTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	history := triggerTask.History.Events
+	scheduledIDs := make(map[string]int64, len(operations))
+	for _, operation := range operations {
+		scheduledIDs[operation] = findNexusScheduledEventID(history, operation)
+		s.Require().Positive(scheduledIDs[operation])
+	}
+	s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{
+		Task: triggerTask,
+		Commands: []*commandpb.Command{
+			requestCancelNexusOperationCommand(findNexusScheduledEventID(history, "cancel-failed")),
+		},
+	})
+	heldTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.Require().NotEmpty(heldTask.TaskToken)
+	close(allowCancellationResponse)
+
+	s.failNexusOperation(ctx, operationCallbacks["failed"])
+	s.cancelBufferedNexusOperation(ctx, operationCallbacks["canceled"])
+	expectedTypes := []enumspb.EventType{
+		enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED,
+		enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED,
+		enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT,
+		enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED,
+	}
+	s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{
+		Namespace:  ns,
+		Execution:  execution,
+		EventTypes: expectedTypes,
+	})
+
+	losingHistory := s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{
+		Namespace:              namespace,
+		Execution:              execution,
+		ReplicationToOldActive: replicationToOldActive,
+		ReplicationToNewActive: replicationToNewActive,
+		ExpectedEventTypes:     expectedTypes,
+		WinnerSignal:           "nexus-outcomes-winner-signal",
+	})
+	for _, eventType := range expectedTypes {
+		s.Require().NotNil(findHistoryEvent(losingHistory, eventType, nil))
+	}
+	winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	for operation, eventType := range map[string]enumspb.EventType{
+		"failed":    enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED,
+		"canceled":  enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED,
+		"timed-out": enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT,
+	} {
+		s.Require().True(
+			hasNexusEventForScheduledID(winningHistory, eventType, scheduledIDs[operation]),
+			"%s for common operation %q must be reapplied to the winning branch",
+			eventType,
+			operation,
+		)
+	}
+	s.Require().False(
+		hasNexusEventForScheduledID(winningHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED, scheduledIDs["cancel-failed"]),
+		"Nexus cancellation results are not cherry-pickable and must remain only on the losing branch",
+	)
+}
+
+// TestNaturallyBufferedNexusCancelRequestCompletedFlushedAndReapplied buffers a successful Nexus
+// cancel-request result. It expects the result to be persisted on the losing branch but skipped on the
+// winner because cancellation results are not cherry-pickable.
+func (s *NexusStateReplicationSuite) TestNaturallyBufferedNexusCancelRequestCompletedFlushedAndReapplied() {
+	if !s.enableTransitionHistory || s.chasmEnabled {
+		s.T().Skip("this conflict-reapplication regression is specific to transition-history HSM Nexus operations")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+	namespace := s.createBufferedEventsNamespace(ctx)
+	ns := namespace.Name
+
+	started := make(chan struct{}, 1)
+	canceled := make(chan struct{}, 1)
+	allowCancellationResponse := make(chan struct{})
+	handler := nexustest.Handler{
+		OnStartOperation: func(_ context.Context, _, operation string, _ *nexus.LazyValue, _ nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
+			started <- struct{}{}
+			return &nexus.HandlerStartOperationResultAsync{OperationToken: operation}, nil
+		},
+		OnCancelOperation: func(_ context.Context, _, _, _ string, _ nexus.CancelOperationOptions) error {
+			canceled <- struct{}{}
+			<-allowCancellationResponse
+			return nil
+		},
+	}
+	endpointName := s.createBufferedNexusEndpoint(ctx, handler)
+	workflowID := "buffered-nexus-cancel-completed-" + uuid.NewString()
+	taskQueue := &taskqueuepb.TaskQueue{Name: workflowID, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
+	execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{
+		Namespace:  ns,
+		WorkflowID: workflowID,
+		TaskQueue:  taskQueue,
+	})
+	firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.completeWorkflowTask(ctx, workflowTaskCompletion{
+		Task:     firstTask,
+		Commands: []*commandpb.Command{scheduleBufferedNexusOperationCommand(endpointName, "cancel-completed")},
+	})
+	select {
+	case <-started:
+	case <-ctx.Done():
+		s.FailNow("timed out waiting for Nexus operation to start")
+	}
+	triggerTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	scheduledID := findNexusScheduledEventID(triggerTask.History.Events, "cancel-completed")
+	s.Require().Positive(scheduledID)
+	s.waitForClusterSynced()
+	replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID)
+	replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID)
+	s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{
+		Task:     triggerTask,
+		Commands: []*commandpb.Command{requestCancelNexusOperationCommand(scheduledID)},
+	})
+	heldTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue)
+	s.Require().NotEmpty(heldTask.TaskToken)
+	close(allowCancellationResponse)
+	select {
+	case <-canceled:
+	case <-ctx.Done():
+		s.FailNow("timed out waiting for Nexus cancellation handler")
+	}
+	expectedTypes := []enumspb.EventType{enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED}
+	s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{
+		Namespace:  ns,
+		Execution:  execution,
+		EventTypes: expectedTypes,
+	})
+
+	s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{
+		Namespace:              namespace,
+		Execution:              execution,
+		ReplicationToOldActive: replicationToOldActive,
+		ReplicationToNewActive: replicationToNewActive,
+		ExpectedEventTypes:     expectedTypes,
+		WinnerSignal:           "nexus-cancel-completed-winner-signal",
+	})
+	winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution)
+	s.Require().False(
+		hasNexusEventForScheduledID(winningHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED, scheduledID),
+		"Nexus cancellation results are not cherry-pickable and must remain only on the losing branch",
+	)
+}