Add version to deletion workflow replication task (#11411)

## What changed? **1. Delete execution replication tasks carry a failover version.** Stamped with the source cluster's failover version at generation (`shard/context_impl.go:1013`), carried in the existing `ReplicationTaskInfo.version` (no proto change), and skipped on apply when older than the target's namespace failover version (`executable_delete_execution_task.go:130`). Still applied: unversioned tasks (queued pre-upgrade), tasks newer than the target's namespace entry, and deletions synthesized by versioned-transition tasks. **2. `DeleteWorkflowExecution` is rejected on a cluster passive for the workflow** (`workflow_handler.go:2513`), with the usual `NamespaceNotActive`. In the frontend, because the history path is shared with replication apply, which must delete on passive clusters — that path, the delete-namespace worker, and admin force-delete are untouched. ## Why? An unversioned delete task generated before a failover kept being applied afterwards, deleting a run the new active cluster owns. ## How did you test it? - [x] built - [x] run locally and tested manually - [x] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) ``` temporal --address :7233 workflow -n global-ns delete -w pay-invoice-0 -r 019fcf5b-996e-7416-9a2d-4ffb0ae7fc13 --grpc-meta xdc-redirection=false WARNING: Deleting Workflow Executions in a global Namespace removes them from all replicas. Requests sent to a passive cluster are forwarded to the active cluster by default; to target the passive cluster directly, specify `--grpc-meta xdc-redirection=false`. Delete Workflow "pay-invoice-0" with Run ID "019fcf5b-996e-7416-9a2d-4ffb0ae7fc13"? y/N y Error: failed to delete workflow: Namespace: global-ns is active in cluster: cluster-b, while current cluster cluster-a is a standby cluster. ``` ## Potential risks - A deletion issued just before a failover is dropped on targets; cleanup waits for the new active cluster's retention timer. Delayed, not leaked — intended trade-off. - API behavior change: deleting against a passive cluster now fails, including batch delete (`batcher/activities.go:704`). No killswitch; admin force-delete is the escape hatch. --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-30 18:41:49 -07:00 · 2026-08-25 15:22:54 -07:00
parent 0e6194c52c
commit a02e33e571
14 changed files with 380 additions and 25 deletions
--- a/common/persistence/serialization/task_serializers.go
+++ b/common/persistence/serialization/task_serializers.go
@@ -1582,6 +1582,7 @@ func replicationDeleteExecutionTaskToProto(
 		TaskId:         task.TaskID,
 		VisibilityTime: timestamppb.New(task.VisibilityTimestamp),
 		ArchetypeId:    task.ArchetypeID,
+		Version:        task.Version,
 	}
 }

@@ -1601,5 +1602,6 @@ func replicationDeleteExecutionTaskFromProto(
 		VisibilityTimestamp: visibilityTimestamp,
 		TaskID:              info.TaskId,
 		ArchetypeID:         info.ArchetypeId,
+		Version:             info.Version,
 	}
 }
--- a/common/persistence/serialization/task_serializers_test.go
+++ b/common/persistence/serialization/task_serializers_test.go
@@ -457,6 +457,7 @@ func (s *taskSerializerSuite) TestDeleteExecutionReplicationTask() {
 		VisibilityTimestamp: time.Unix(0, 0).UTC(), // go == compare for location as well which is striped during marshaling/unmarshaling
 		TaskID:              rand.Int63(),
 		ArchetypeID:         rand.Uint32(),
+		Version:             rand.Int63(),
 	}

 	s.assertEqualTasks(deleteExecutionReplicationTask)
--- a/service/frontend/workflow_handler.go
+++ b/service/frontend/workflow_handler.go
@@ -2518,13 +2518,17 @@ func (wh *WorkflowHandler) DeleteWorkflowExecution(ctx context.Context, request
 		return nil, err
 	}

-	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
+	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
 	if err != nil {
 		return nil, err
 	}

+	if err := wh.validateWorkflowDeletionCluster(namespaceEntry, request.GetWorkflowExecution().GetWorkflowId()); err != nil {
+		return nil, err
+	}
+
 	_, err = wh.historyClient.DeleteWorkflowExecution(ctx, &historyservice.DeleteWorkflowExecutionRequest{
-		NamespaceId:        namespaceID.String(),
+		NamespaceId:        namespaceEntry.ID().String(),
 		WorkflowExecution:  request.GetWorkflowExecution(),
 		ClosedWorkflowOnly: false,
 	})
@@ -2535,6 +2539,35 @@ func (wh *WorkflowHandler) DeleteWorkflowExecution(ctx context.Context, request
 	return &workflowservice.DeleteWorkflowExecutionResponse{}, nil
 }

+// validateWorkflowDeletionCluster rejects a deletion that targets a cluster which is passive for the
+// workflow. A deletion performed on a passive cluster is not replicated: it only drops the local copy
+// while the active cluster still holds the execution and keeps replicating it back, so the two
+// clusters diverge (and the local copy can even be resurrected by a later replication task). The
+// caller must delete on the active cluster, which replicates the deletion to every other cluster.
+//
+// When XDC redirection is enabled the request is forwarded to the active cluster before it gets here,
+// so this only rejects requests that would otherwise be served locally on a passive cluster. Deleting
+// local state on a passive cluster is still possible through the admin ForceDeleteWorkflowExecution
+// API, which does not go through this handler.
+func (wh *WorkflowHandler) validateWorkflowDeletionCluster(
+	namespaceEntry *namespace.Namespace,
+	workflowID string,
+) error {
+	if !namespaceEntry.IsGlobalNamespace() {
+		return nil
+	}
+	currentCluster := wh.clusterMetadata.GetCurrentClusterName()
+	activeCluster := namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: workflowID})
+	if activeCluster == currentCluster {
+		return nil
+	}
+	return serviceerror.NewNamespaceNotActive(
+		namespaceEntry.Name().String(),
+		currentCluster,
+		activeCluster,
+	)
+}
+
 // ListOpenWorkflowExecutions is a visibility API to list the open executions in a specific namespace.
 func (wh *WorkflowHandler) ListOpenWorkflowExecutions(ctx context.Context, request *workflowservice.ListOpenWorkflowExecutionsRequest) (_ *workflowservice.ListOpenWorkflowExecutionsResponse, retError error) {
 	defer log.CapturePanic(wh.logger, &retError)
--- a/service/frontend/workflow_handler_test.go
+++ b/service/frontend/workflow_handler_test.go
@@ -4696,7 +4696,8 @@ func (s *WorkflowHandlerSuite) Test_DeleteWorkflowExecution() {

 	// History call failed.
 	s.mockResource.HistoryClient.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, errors.New("random error"))
-	s.mockResource.NamespaceCache.EXPECT().GetNamespaceID(namespace.Name("test-namespace")).Return(namespace.ID("test-namespace-id"), nil)
+	s.mockResource.NamespaceCache.EXPECT().GetNamespace(namespace.Name("test-namespace")).
+		Return(s.localNamespaceEntry("test-namespace", "test-namespace-id"), nil)
 	resp, err := wh.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
 		Namespace: "test-namespace",
 		WorkflowExecution: &commonpb.WorkflowExecution{
@@ -4710,7 +4711,8 @@ func (s *WorkflowHandlerSuite) Test_DeleteWorkflowExecution() {

 	// Success case.
 	s.mockResource.HistoryClient.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(&historyservice.DeleteWorkflowExecutionResponse{}, nil)
-	s.mockResource.NamespaceCache.EXPECT().GetNamespaceID(namespace.Name("test-namespace")).Return(namespace.ID("test-namespace-id"), nil)
+	s.mockResource.NamespaceCache.EXPECT().GetNamespace(namespace.Name("test-namespace")).
+		Return(s.localNamespaceEntry("test-namespace", "test-namespace-id"), nil)
 	resp, err = wh.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
 		Namespace: "test-namespace",
 		WorkflowExecution: &commonpb.WorkflowExecution{
@@ -4722,6 +4724,69 @@ func (s *WorkflowHandlerSuite) Test_DeleteWorkflowExecution() {
 	s.NotNil(resp)
 }

+// A deletion is only replicated when it happens on the active cluster, so a request that lands on a
+// passive cluster (XDC redirection disabled) must be rejected instead of deleting the local copy only.
+func (s *WorkflowHandlerSuite) Test_DeleteWorkflowExecution_PassiveCluster() {
+	wh := s.getWorkflowHandler(s.newConfig())
+
+	s.mockResource.NamespaceCache.EXPECT().GetNamespace(namespace.Name("test-namespace")).
+		Return(s.globalNamespaceEntry("test-namespace", "test-namespace-id", cluster.TestAlternativeClusterName), nil)
+	s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
+	// No DeleteWorkflowExecution call is expected on the history client.
+
+	resp, err := wh.DeleteWorkflowExecution(context.Background(), &workflowservice.DeleteWorkflowExecutionRequest{
+		Namespace: "test-namespace",
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: "test-workflow-id",
+		},
+	})
+	s.Nil(resp)
+	var notActiveErr *serviceerror.NamespaceNotActive
+	s.ErrorAs(err, &notActiveErr)
+}
+
+func (s *WorkflowHandlerSuite) Test_DeleteWorkflowExecution_ActiveCluster() {
+	wh := s.getWorkflowHandler(s.newConfig())
+
+	s.mockResource.NamespaceCache.EXPECT().GetNamespace(namespace.Name("test-namespace")).
+		Return(s.globalNamespaceEntry("test-namespace", "test-namespace-id", cluster.TestCurrentClusterName), nil)
+	s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
+	s.mockResource.HistoryClient.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).
+		Return(&historyservice.DeleteWorkflowExecutionResponse{}, nil)
+
+	resp, err := wh.DeleteWorkflowExecution(context.Background(), &workflowservice.DeleteWorkflowExecutionRequest{
+		Namespace: "test-namespace",
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: "test-workflow-id",
+		},
+	})
+	s.NoError(err)
+	s.NotNil(resp)
+}
+
+func (s *WorkflowHandlerSuite) localNamespaceEntry(nsName string, nsID string) *namespace.Namespace {
+	return namespace.NewLocalNamespaceForTest(
+		&persistencespb.NamespaceInfo{Id: nsID, Name: nsName},
+		nil,
+		cluster.TestCurrentClusterName,
+	)
+}
+
+func (s *WorkflowHandlerSuite) globalNamespaceEntry(nsName string, nsID string, activeCluster string) *namespace.Namespace {
+	return namespace.NewGlobalNamespaceForTest(
+		&persistencespb.NamespaceInfo{Id: nsID, Name: nsName},
+		nil,
+		&persistencespb.NamespaceReplicationConfig{
+			ActiveClusterName: activeCluster,
+			Clusters: []string{
+				cluster.TestCurrentClusterName,
+				cluster.TestAlternativeClusterName,
+			},
+		},
+		cluster.TestCurrentClusterInitialFailoverVersion,
+	)
+}
+
 func (s *WorkflowHandlerSuite) TestExecuteMultiOperation() {
 	ctx := context.Background()
 	config := s.newConfig()
--- a/service/history/deletemanager/delete_manager.go
+++ b/service/history/deletemanager/delete_manager.go
@@ -6,6 +6,7 @@ import (
 	"context"

 	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/metrics"
@@ -181,6 +182,13 @@ func (m *DeleteManagerImpl) deleteWorkflowExecutionInternal(
 	}

 	executionInfo := ms.GetExecutionInfo()
+	lastWriteVersion := common.EmptyVersion
+	if !retentionDelete {
+		lastWriteVersion, err = ms.GetLastWriteVersion()
+		if err != nil {
+			return err
+		}
+	}
 	if err := m.shardContext.DeleteWorkflowExecution(
 		ctx,
 		definition.WorkflowKey{
@@ -189,6 +197,7 @@ func (m *DeleteManagerImpl) deleteWorkflowExecutionInternal(
 			RunID:       we.GetRunId(),
 		},
 		ms.ChasmTree().ArchetypeID(),
+		lastWriteVersion,
 		currentBranchToken,
 		executionInfo.GetCloseVisibilityTaskId(),
 		closeTime,
--- a/service/history/deletemanager/delete_manager_test.go
+++ b/service/history/deletemanager/delete_manager_test.go
@@ -11,6 +11,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/definition"
@@ -96,6 +97,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution() {
 	})
 	mockMutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{})
 	mockMutableState.EXPECT().GetWorkflowCloseTime(gomock.Any()).Return(time.Unix(0, 0).UTC(), nil)
+	mockMutableState.EXPECT().GetLastWriteVersion().Return(tests.Version, nil)
 	mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
 	stage := tasks.DeleteWorkflowExecutionStageNone

@@ -107,6 +109,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution() {
 			RunID:       tests.RunID,
 		},
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		[]byte{22, 8, 78},
 		closeExecutionVisibilityTaskID,
 		time.Unix(0, 0).UTC(),
@@ -142,6 +145,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution_Error()
 	})
 	mockMutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{})
 	mockMutableState.EXPECT().GetWorkflowCloseTime(gomock.Any()).Return(time.Unix(0, 0).UTC(), nil)
+	mockMutableState.EXPECT().GetLastWriteVersion().Return(tests.Version, nil)
 	mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
 	stage := tasks.DeleteWorkflowExecutionStageNone

@@ -153,6 +157,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution_Error()
 			RunID:       tests.RunID,
 		},
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		[]byte{22, 8, 78},
 		closeExecutionVisibilityTaskID,
 		time.Unix(0, 0).UTC(),
@@ -198,6 +203,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecutionByRetention_Skip
 			RunID:       tests.RunID,
 		},
 		workflow.NoopChasmTree.ArchetypeID(),
+		common.EmptyVersion,
 		[]byte{22, 8, 78},
 		closeExecutionVisibilityTaskID,
 		time.Unix(0, 0).UTC(),
@@ -208,6 +214,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecutionByRetention_Skip
 		_ context.Context,
 		_ definition.WorkflowKey,
 		_ chasm.ArchetypeID,
+		_ int64,
 		_ []byte,
 		_ int64,
 		_ time.Time,
@@ -248,6 +255,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecution_OpenWorkflow()
 	})
 	mockMutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{})
 	mockMutableState.EXPECT().GetWorkflowCloseTime(gomock.Any()).Return(time.Unix(0, 0).UTC(), nil)
+	mockMutableState.EXPECT().GetLastWriteVersion().Return(tests.Version, nil)
 	mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
 	stage := tasks.DeleteWorkflowExecutionStageNone

@@ -259,6 +267,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecution_OpenWorkflow()
 			RunID:       tests.RunID,
 		},
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		[]byte{22, 8, 78},
 		closeExecutionVisibilityTaskID,
 		time.Unix(0, 0).UTC(),
--- a/service/history/interfaces/shard_context.go
+++ b/service/history/interfaces/shard_context.go
@@ -105,6 +105,7 @@ type (
 			ctx context.Context,
 			workflowKey definition.WorkflowKey,
 			archetypeID chasm.ArchetypeID,
+			lastWriteVersion int64,
 			branchToken []byte,
 			closeExecutionVisibilityTaskID int64,
 			workflowCloseTime time.Time,
--- a/service/history/interfaces/shard_context_mock.go
+++ b/service/history/interfaces/shard_context_mock.go
@@ -211,17 +211,17 @@ func (mr *MockShardContextMockRecorder) CurrentVectorClock() *gomock.Call {
 }

 // DeleteWorkflowExecution mocks base method.
-func (m *MockShardContext) DeleteWorkflowExecution(ctx context.Context, workflowKey definition.WorkflowKey, archetypeID chasm.ArchetypeID, branchToken []byte, closeExecutionVisibilityTaskID int64, workflowCloseTime, workflowStartTime time.Time, stage *tasks.DeleteWorkflowExecutionStage, retentionDelete bool) error {
+func (m *MockShardContext) DeleteWorkflowExecution(ctx context.Context, workflowKey definition.WorkflowKey, archetypeID chasm.ArchetypeID, lastWriteVersion int64, branchToken []byte, closeExecutionVisibilityTaskID int64, workflowCloseTime, workflowStartTime time.Time, stage *tasks.DeleteWorkflowExecutionStage, retentionDelete bool) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DeleteWorkflowExecution", ctx, workflowKey, archetypeID, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
+	ret := m.ctrl.Call(m, "DeleteWorkflowExecution", ctx, workflowKey, archetypeID, lastWriteVersion, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
 	ret0, _ := ret[0].(error)
 	return ret0
 }

 // DeleteWorkflowExecution indicates an expected call of DeleteWorkflowExecution.
-func (mr *MockShardContextMockRecorder) DeleteWorkflowExecution(ctx, workflowKey, archetypeID, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete any) *gomock.Call {
+func (mr *MockShardContextMockRecorder) DeleteWorkflowExecution(ctx, workflowKey, archetypeID, lastWriteVersion, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteWorkflowExecution", reflect.TypeOf((*MockShardContext)(nil).DeleteWorkflowExecution), ctx, workflowKey, archetypeID, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteWorkflowExecution", reflect.TypeOf((*MockShardContext)(nil).DeleteWorkflowExecution), ctx, workflowKey, archetypeID, lastWriteVersion, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
 }

 // EndpointRegistry mocks base method.
@@ -1058,17 +1058,17 @@ func (mr *MockControllableContextMockRecorder) CurrentVectorClock() *gomock.Call
 }

 // DeleteWorkflowExecution mocks base method.
-func (m *MockControllableContext) DeleteWorkflowExecution(ctx context.Context, workflowKey definition.WorkflowKey, archetypeID chasm.ArchetypeID, branchToken []byte, closeExecutionVisibilityTaskID int64, workflowCloseTime, workflowStartTime time.Time, stage *tasks.DeleteWorkflowExecutionStage, retentionDelete bool) error {
+func (m *MockControllableContext) DeleteWorkflowExecution(ctx context.Context, workflowKey definition.WorkflowKey, archetypeID chasm.ArchetypeID, lastWriteVersion int64, branchToken []byte, closeExecutionVisibilityTaskID int64, workflowCloseTime, workflowStartTime time.Time, stage *tasks.DeleteWorkflowExecutionStage, retentionDelete bool) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DeleteWorkflowExecution", ctx, workflowKey, archetypeID, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
+	ret := m.ctrl.Call(m, "DeleteWorkflowExecution", ctx, workflowKey, archetypeID, lastWriteVersion, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
 	ret0, _ := ret[0].(error)
 	return ret0
 }

 // DeleteWorkflowExecution indicates an expected call of DeleteWorkflowExecution.
-func (mr *MockControllableContextMockRecorder) DeleteWorkflowExecution(ctx, workflowKey, archetypeID, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete any) *gomock.Call {
+func (mr *MockControllableContextMockRecorder) DeleteWorkflowExecution(ctx, workflowKey, archetypeID, lastWriteVersion, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteWorkflowExecution", reflect.TypeOf((*MockControllableContext)(nil).DeleteWorkflowExecution), ctx, workflowKey, archetypeID, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteWorkflowExecution", reflect.TypeOf((*MockControllableContext)(nil).DeleteWorkflowExecution), ctx, workflowKey, archetypeID, lastWriteVersion, branchToken, closeExecutionVisibilityTaskID, workflowCloseTime, workflowStartTime, stage, retentionDelete)
 }

 // EndpointRegistry mocks base method.
--- a/service/history/replication/executable_delete_execution_task.go
+++ b/service/history/replication/executable_delete_execution_task.go
@@ -12,8 +12,10 @@ import (
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
 	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/headers"
+	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
@@ -27,6 +29,11 @@ type ExecutableDeleteExecutionTask struct {

 	chasm.ComponentRef
 	ExecutableTask
+
+	// lastWriteVersion is the source execution's last write version when it was deleted.
+	// It is common.EmptyVersion when the source did not stamp one, i.e. for tasks generated before
+	// the version was introduced and for deletions synthesized from another replication task.
+	lastWriteVersion int64
 }

 var _ ctasks.Task = (*ExecutableDeleteExecutionTask)(nil)
@@ -50,6 +57,14 @@ func NewExecutableDeleteExecutionTask(
 		softassert.That(processToolBox.Logger, false, "delete execution replication task has unspecified archetype ID")
 	}

+	// Only take the version from a genuine delete execution replication task. Other replication
+	// tasks (sync/verify versioned transition) synthesize a deletion out of their own task, whose
+	// version describes a different operation and must not be interpreted as a deletion version.
+	lastWriteVersion := common.EmptyVersion
+	if rawInfo.GetTaskType() == enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION {
+		lastWriteVersion = rawInfo.GetVersion()
+	}
+
 	return &ExecutableDeleteExecutionTask{
 		ProcessToolBox: processToolBox,
 		ComponentRef: chasm.NewComponentRefByArchetypeID(
@@ -70,6 +85,7 @@ func NewExecutableDeleteExecutionTask(
 			sourceShardKey,
 			replicationTask,
 		),
+		lastWriteVersion: lastWriteVersion,
 	}
 }

@@ -115,7 +131,9 @@ func (e *ExecutableDeleteExecutionTask) Execute() error {
 		return err
 	}
 	currentCluster := e.ClusterMetadata.GetCurrentClusterName()
-	if namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: e.BusinessID}) == currentCluster {
+	// Legacy tasks have no execution-state fence and remain unsafe to apply on an active cluster.
+	if e.lastWriteVersion == common.EmptyVersion &&
+		namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: e.BusinessID}) == currentCluster {
 		e.Logger.Warn("Skipping delete execution replication task on active cluster",
 			tag.WorkflowNamespaceID(e.NamespaceID),
 			tag.WorkflowID(e.BusinessID),
@@ -138,6 +156,28 @@ func (e *ExecutableDeleteExecutionTask) Execute() error {
 	if err != nil {
 		return err
 	}
+	if e.lastWriteVersion != common.EmptyVersion {
+		targetLastWriteVersion, err := e.getLastWriteVersion(ctx, archetypeID)
+		if err != nil {
+			return err
+		}
+		if e.lastWriteVersion != targetLastWriteVersion {
+			e.Logger.Warn("Skipping delete execution replication task due to last write version mismatch",
+				tag.WorkflowNamespaceID(e.NamespaceID),
+				tag.WorkflowID(e.BusinessID),
+				tag.WorkflowRunID(e.RunID),
+				tag.TaskID(e.TaskID()),
+				tag.IncomingVersion(e.lastWriteVersion),
+				tag.CurrentVersion(targetLastWriteVersion),
+			)
+			metrics.ReplicationTasksSkipped.With(e.MetricsHandler).Record(
+				1,
+				metrics.OperationTag(metrics.DeleteExecutionReplicationTaskScope),
+				metrics.NamespaceTag(namespaceName),
+			)
+			return nil
+		}
+	}
 	switch archetypeID {
 	case chasm.WorkflowArchetypeID:
 		return e.deleteWorkflowExecution(ctx)
@@ -173,6 +213,38 @@ func (e *ExecutableDeleteExecutionTask) deleteChasmExecution(ctx context.Context
 	return e.ChasmEngine.DeleteExecution(ctx, e.ComponentRef, chasm.DeleteExecutionRequest{})
 }

+func (e *ExecutableDeleteExecutionTask) getLastWriteVersion(
+	ctx context.Context,
+	archetypeID chasm.ArchetypeID,
+) (_ int64, retError error) {
+	namespaceID := namespace.ID(e.NamespaceID)
+	shardContext, err := e.ShardController.GetShardByNamespaceWorkflow(namespaceID, e.BusinessID)
+	if err != nil {
+		return common.EmptyVersion, err
+	}
+	workflowContext, release, err := e.WorkflowCache.GetOrCreateChasmExecution(
+		ctx,
+		shardContext,
+		namespaceID,
+		&commonpb.WorkflowExecution{
+			WorkflowId: e.BusinessID,
+			RunId:      e.RunID,
+		},
+		archetypeID,
+		locks.PriorityLow,
+	)
+	if err != nil {
+		return common.EmptyVersion, err
+	}
+	defer func() { release(retError) }()
+
+	mutableState, err := workflowContext.LoadMutableState(ctx, shardContext)
+	if err != nil {
+		return common.EmptyVersion, err
+	}
+	return mutableState.GetLastWriteVersion()
+}
+
 func (e *ExecutableDeleteExecutionTask) HandleErr(err error) error {
 	metrics.ReplicationTasksErrorByType.With(e.MetricsHandler).Record(
 		1,
--- a/service/history/replication/executable_delete_execution_task_test.go
+++ b/service/history/replication/executable_delete_execution_task_test.go
@@ -14,7 +14,9 @@ import (
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
 	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/cluster"
+	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
@@ -22,6 +24,7 @@ import (
 	historyi "go.temporal.io/server/service/history/interfaces"
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/tests"
+	wcache "go.temporal.io/server/service/history/workflow/cache"
 	"go.uber.org/mock/gomock"
 )

@@ -34,6 +37,8 @@ type (
 		clusterMetadata *cluster.MockMetadata
 		shardController *shard.MockController
 		namespaceCache  *namespace.MockRegistry
+		workflowCache   *wcache.MockCache
+		chasmEngine     *chasm.MockEngine
 		metricsHandler  metrics.Handler
 		logger          log.Logger
 		config          *configs.Config
@@ -63,6 +68,8 @@ func (s *executableDeleteExecutionTaskSuite) SetupTest() {
 	s.clusterMetadata = cluster.NewMockMetadata(s.controller)
 	s.shardController = shard.NewMockController(s.controller)
 	s.namespaceCache = namespace.NewMockRegistry(s.controller)
+	s.workflowCache = wcache.NewMockCache(s.controller)
+	s.chasmEngine = chasm.NewMockEngine(s.controller)
 	s.metricsHandler = metrics.NoopMetricsHandler
 	s.logger = log.NewNoopLogger()
 	s.config = tests.NewDynamicConfig()
@@ -81,10 +88,12 @@ func (s *executableDeleteExecutionTaskSuite) SetupTest() {
 		Config:          s.config,
 		ClusterMetadata: s.clusterMetadata,
 		ShardController: s.shardController,
+		ChasmEngine:     s.chasmEngine,
 		NamespaceCache:  s.namespaceCache,
 		MetricsHandler:  s.metricsHandler,
 		Logger:          s.logger,
 		ThrottledLogger: s.logger,
+		WorkflowCache:   s.workflowCache,
 	}

 	s.clusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
@@ -105,6 +114,116 @@ func (s *executableDeleteExecutionTaskSuite) TestExecute_CurrentClusterPassive_D
 	namespaceEntry := s.namespaceEntry(cluster.TestAlternativeClusterName)
 	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)

+	s.expectDeleteWorkflowExecution()
+
+	s.NoError(s.newTask().Execute())
+}
+
+func (s *executableDeleteExecutionTaskSuite) TestExecute_MatchingLastWriteVersion_DeletesWorkflowExecution() {
+	lastWriteVersion := cluster.TestAlternativeClusterInitialFailoverVersion
+	namespaceEntry := s.namespaceEntry(cluster.TestAlternativeClusterName)
+	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)
+
+	s.expectLastWriteVersion(chasm.WorkflowArchetypeID, lastWriteVersion)
+	s.expectDeleteWorkflowExecution()
+
+	s.NoError(s.newTaskWithVersion(lastWriteVersion).Execute())
+}
+
+func (s *executableDeleteExecutionTaskSuite) TestExecute_MismatchedLastWriteVersion_SkipsTask() {
+	lastWriteVersion := cluster.TestAlternativeClusterInitialFailoverVersion
+	namespaceEntry := s.namespaceEntry(cluster.TestAlternativeClusterName)
+	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)
+
+	s.expectLastWriteVersion(chasm.WorkflowArchetypeID, lastWriteVersion+cluster.TestFailoverVersionIncrement)
+
+	s.NoError(s.newTaskWithVersion(lastWriteVersion).Execute())
+}
+
+func (s *executableDeleteExecutionTaskSuite) TestExecute_VersionedTaskOnActiveCluster_DeletesWorkflowExecution() {
+	lastWriteVersion := cluster.TestAlternativeClusterInitialFailoverVersion
+	namespaceEntry := s.namespaceEntry(cluster.TestCurrentClusterName)
+	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)
+
+	s.expectLastWriteVersion(chasm.WorkflowArchetypeID, lastWriteVersion)
+	s.expectDeleteWorkflowExecution()
+
+	s.NoError(s.newTaskWithVersion(lastWriteVersion).Execute())
+}
+
+func (s *executableDeleteExecutionTaskSuite) TestExecute_MatchingLastWriteVersion_DeletesChasmExecution() {
+	lastWriteVersion := cluster.TestAlternativeClusterInitialFailoverVersion
+	archetypeID := chasm.ArchetypeID(42)
+	namespaceEntry := s.namespaceEntry(cluster.TestAlternativeClusterName)
+	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)
+
+	s.expectLastWriteVersion(archetypeID, lastWriteVersion)
+	s.chasmEngine.EXPECT().DeleteExecution(gomock.Any(), gomock.Any(), chasm.DeleteExecutionRequest{}).Return(nil)
+
+	s.NoError(s.newTaskWithVersionAndArchetype(lastWriteVersion, archetypeID).Execute())
+}
+
+// Tasks generated before the version was introduced carry no version and must keep being applied.
+func (s *executableDeleteExecutionTaskSuite) TestExecute_UnversionedTask_DeletesWorkflowExecution() {
+	namespaceEntry := s.namespaceEntry(cluster.TestAlternativeClusterName)
+	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)
+
+	s.expectDeleteWorkflowExecution()
+
+	s.NoError(s.newTaskWithVersion(common.EmptyVersion).Execute())
+}
+
+// Sync/verify versioned transition tasks synthesize a deletion out of their own replication task.
+// Their version describes a different operation and must not be treated as a deletion version.
+func (s *executableDeleteExecutionTaskSuite) TestExecute_SynthesizedDeletion_IgnoresSourceTaskVersion() {
+	namespaceEntry := s.namespaceEntry(cluster.TestAlternativeClusterName)
+	s.namespaceCache.EXPECT().GetNamespaceByID(s.namespaceID).Return(namespaceEntry, nil).Times(3)
+
+	s.expectDeleteWorkflowExecution()
+
+	task := NewExecutableDeleteExecutionTask(
+		s.processToolBox,
+		s.taskID,
+		s.taskCreationTime,
+		s.sourceClusterName,
+		s.sourceShardKey,
+		&replicationspb.ReplicationTask{
+			TaskType: enumsspb.REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK,
+			RawTaskInfo: &persistencespb.ReplicationTaskInfo{
+				NamespaceId: s.namespaceID.String(),
+				WorkflowId:  s.workflowID,
+				RunId:       s.runID,
+				TaskId:      s.taskID,
+				TaskType:    enumsspb.TASK_TYPE_REPLICATION_SYNC_VERSIONED_TRANSITION,
+				ArchetypeId: chasm.WorkflowArchetypeID,
+				Version:     cluster.TestAlternativeClusterInitialFailoverVersion,
+			},
+		},
+	)
+	s.NoError(task.Execute())
+}
+
+func (s *executableDeleteExecutionTaskSuite) expectLastWriteVersion(
+	archetypeID chasm.ArchetypeID,
+	lastWriteVersion int64,
+) {
+	shardContext := historyi.NewMockShardContext(s.controller)
+	workflowContext := historyi.NewMockWorkflowContext(s.controller)
+	mutableState := historyi.NewMockMutableState(s.controller)
+	s.shardController.EXPECT().GetShardByNamespaceWorkflow(s.namespaceID, s.workflowID).Return(shardContext, nil)
+	s.workflowCache.EXPECT().GetOrCreateChasmExecution(
+		gomock.Any(),
+		shardContext,
+		s.namespaceID,
+		&commonpb.WorkflowExecution{WorkflowId: s.workflowID, RunId: s.runID},
+		archetypeID,
+		locks.PriorityLow,
+	).Return(workflowContext, func(error) {}, nil)
+	workflowContext.EXPECT().LoadMutableState(gomock.Any(), shardContext).Return(mutableState, nil)
+	mutableState.EXPECT().GetLastWriteVersion().Return(lastWriteVersion, nil)
+}
+
+func (s *executableDeleteExecutionTaskSuite) expectDeleteWorkflowExecution() {
 	shardContext := historyi.NewMockShardContext(s.controller)
 	engine := historyi.NewMockEngine(s.controller)
 	s.shardController.EXPECT().GetShardByNamespaceWorkflow(s.namespaceID, s.workflowID).Return(shardContext, nil)
@@ -116,11 +235,20 @@ func (s *executableDeleteExecutionTaskSuite) TestExecute_CurrentClusterPassive_D
 			RunId:      s.runID,
 		},
 	}).Return(&historyservice.DeleteWorkflowExecutionResponse{}, nil)
-
-	s.NoError(s.newTask().Execute())
 }

 func (s *executableDeleteExecutionTaskSuite) newTask() *ExecutableDeleteExecutionTask {
+	return s.newTaskWithVersion(common.EmptyVersion)
+}
+
+func (s *executableDeleteExecutionTaskSuite) newTaskWithVersion(version int64) *ExecutableDeleteExecutionTask {
+	return s.newTaskWithVersionAndArchetype(version, chasm.WorkflowArchetypeID)
+}
+
+func (s *executableDeleteExecutionTaskSuite) newTaskWithVersionAndArchetype(
+	version int64,
+	archetypeID chasm.ArchetypeID,
+) *ExecutableDeleteExecutionTask {
 	return NewExecutableDeleteExecutionTask(
 		s.processToolBox,
 		s.taskID,
@@ -134,13 +262,22 @@ func (s *executableDeleteExecutionTaskSuite) newTask() *ExecutableDeleteExecutio
 				WorkflowId:  s.workflowID,
 				RunId:       s.runID,
 				TaskId:      s.taskID,
-				ArchetypeId: chasm.WorkflowArchetypeID,
+				TaskType:    enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION,
+				ArchetypeId: archetypeID,
+				Version:     version,
 			},
 		},
 	)
 }

 func (s *executableDeleteExecutionTaskSuite) namespaceEntry(activeCluster string) *namespace.Namespace {
+	return s.namespaceEntryWithFailoverVersion(activeCluster, cluster.TestCurrentClusterInitialFailoverVersion)
+}
+
+func (s *executableDeleteExecutionTaskSuite) namespaceEntryWithFailoverVersion(
+	activeCluster string,
+	failoverVersion int64,
+) *namespace.Namespace {
 	detail := &persistencespb.NamespaceDetail{
 		Info: &persistencespb.NamespaceInfo{
 			Id:   s.namespaceID.String(),
@@ -154,7 +291,7 @@ func (s *executableDeleteExecutionTaskSuite) namespaceEntry(activeCluster string
 				cluster.TestAlternativeClusterName,
 			},
 		},
-		FailoverVersion: cluster.TestCurrentClusterInitialFailoverVersion,
+		FailoverVersion: failoverVersion,
 	}
 	namespaceEntry, err := namespace.FromPersistentState(
 		detail,
--- a/service/history/shard/context_impl.go
+++ b/service/history/shard/context_impl.go
@@ -911,6 +911,7 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 	ctx context.Context,
 	key definition.WorkflowKey,
 	archetypeID chasm.ArchetypeID,
+	lastWriteVersion int64,
 	branchToken []byte,
 	closeVisibilityTaskId int64,
 	workflowCloseTime time.Time,
@@ -1010,6 +1011,7 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 							&tasks.DeleteExecutionReplicationTask{
 								WorkflowKey: key,
 								ArchetypeID: archetypeID,
+								Version:     lastWriteVersion,
 							},
 						}
 					}
--- a/service/history/shard/context_test.go
+++ b/service/history/shard/context_test.go
@@ -208,6 +208,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -236,6 +237,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -253,6 +255,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -269,6 +272,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -296,6 +300,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -312,6 +317,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -328,6 +334,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -343,6 +350,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -355,7 +363,8 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 }

 func (s *contextSuite) TestDeleteWorkflowExecution_EmitsReplicationTaskWhenWorkflowActiveInCurrentCluster() {
-	captured := s.runDeleteWorkflowExecutionForReplicationCheck(cluster.TestCurrentClusterName)
+	lastWriteVersion := tests.Version + 100
+	captured := s.runDeleteWorkflowExecutionForReplicationCheck(cluster.TestCurrentClusterName, lastWriteVersion)

 	replicationTasks := captured.Tasks[tasks.CategoryReplication]
 	s.Require().Len(replicationTasks, 1, "expected a DeleteExecutionReplicationTask when workflow is active in current cluster")
@@ -363,10 +372,11 @@ func (s *contextSuite) TestDeleteWorkflowExecution_EmitsReplicationTaskWhenWorkf
 	s.True(ok, "task should be *DeleteExecutionReplicationTask")
 	s.Equal(captured.WorkflowID, deleteTask.WorkflowID)
 	s.Equal(captured.NamespaceID, deleteTask.NamespaceID)
+	s.Equal(lastWriteVersion, deleteTask.Version)
 }

 func (s *contextSuite) TestDeleteWorkflowExecution_NoReplicationTaskWhenWorkflowActiveInOtherCluster() {
-	captured := s.runDeleteWorkflowExecutionForReplicationCheck(cluster.TestAlternativeClusterName)
+	captured := s.runDeleteWorkflowExecutionForReplicationCheck(cluster.TestAlternativeClusterName, tests.Version)

 	s.Empty(captured.Tasks[tasks.CategoryReplication],
 		"expected no DeleteExecutionReplicationTask when workflow is active in another cluster")
@@ -374,6 +384,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_NoReplicationTaskWhenWorkflow

 func (s *contextSuite) runDeleteWorkflowExecutionForReplicationCheck(
 	workflowActiveCluster string,
+	lastWriteVersion int64,
 ) *persistence.AddHistoryTasksRequest {
 	nsID := namespace.NewID()
 	nsEntry := namespace.NewGlobalNamespaceForTest(
@@ -412,6 +423,7 @@ func (s *contextSuite) runDeleteWorkflowExecutionForReplicationCheck(
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		lastWriteVersion,
 		[]byte("branchToken"),
 		0,
 		time.Time{},
@@ -439,6 +451,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_DeleteVisibilityTaskNotificti
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
@@ -457,6 +470,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_DeleteVisibilityTaskNotificti
 		context.Background(),
 		workflowKey,
 		chasm.WorkflowArchetypeID,
+		tests.Version,
 		branchToken,
 		0,
 		time.Time{},
--- a/service/history/tasks/delete_execution_replication_task.go
+++ b/service/history/tasks/delete_execution_replication_task.go
@@ -15,6 +15,10 @@ type DeleteExecutionReplicationTask struct {
 	VisibilityTimestamp time.Time
 	TaskID              int64
 	ArchetypeID         uint32
+	// Version is the namespace failover version of the cluster that deleted the execution.
+	// The target cluster uses it to drop deletions that were generated before a failover.
+	// It is common.EmptyVersion for tasks generated before this field was introduced.
+	Version int64
 }

 func (a *DeleteExecutionReplicationTask) GetKey() Key {
@@ -22,10 +26,11 @@ func (a *DeleteExecutionReplicationTask) GetKey() Key {
 }

 func (a *DeleteExecutionReplicationTask) GetVersion() int64 {
-	return 0
+	return a.Version
 }

-func (a *DeleteExecutionReplicationTask) SetVersion(_ int64) {
+func (a *DeleteExecutionReplicationTask) SetVersion(version int64) {
+	a.Version = version
 }

 func (a *DeleteExecutionReplicationTask) GetTaskID() int64 {
--- a/tests/xdc/stream_based_replication_test.go
+++ b/tests/xdc/stream_based_replication_test.go
@@ -473,17 +473,22 @@ func (s *streamBasedReplicationTestSuite) TestForceReplicateResetWorkflow_BaseWo
 	})
 	s.NoError(err)

+	// Wipe the local copies on the passive cluster. The frontend DeleteWorkflowExecution API rejects
+	// deletions on a cluster that is passive for the workflow (they would not be replicated), so go
+	// through the history service directly, which is the same path replication apply uses. The admin
+	// force-delete API is not usable here: it deletes the DB rows without clearing the workflow cache,
+	// so the target keeps serving the deleted runs from cache.
 	client1 := s.clusters[1].FrontendClient()
-	_, err = client1.DeleteWorkflowExecution(testcore.NewContext(), &workflowservice.DeleteWorkflowExecutionRequest{
-		Namespace: ns,
+	_, err = s.clusters[1].HistoryClient().DeleteWorkflowExecution(testcore.NewContext(), &historyservice.DeleteWorkflowExecutionRequest{
+		NamespaceId: resp.NamespaceInfo.GetId(),
 		WorkflowExecution: &commonpb.WorkflowExecution{
 			WorkflowId: id,
 			RunId:      we.GetRunId(),
 		},
 	})
 	s.NoError(err)
-	_, err = client1.DeleteWorkflowExecution(testcore.NewContext(), &workflowservice.DeleteWorkflowExecutionRequest{
-		Namespace: ns,
+	_, err = s.clusters[1].HistoryClient().DeleteWorkflowExecution(testcore.NewContext(), &historyservice.DeleteWorkflowExecutionRequest{
+		NamespaceId: resp.NamespaceInfo.GetId(),
 		WorkflowExecution: &commonpb.WorkflowExecution{
 			WorkflowId: id,
 			RunId:      resetResp.GetRunId(),