diff --git a/test/s3/acl/s3_acl_versioning_test.go b/test/s3/acl/s3_acl_versioning_test.go
new file mode 100644
index 000000000..7e7326ec4
--- /dev/null
+++ b/test/s3/acl/s3_acl_versioning_test.go
@@ -0,0 +1,335 @@
+package acl
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func getS3Client(t *testing.T) *s3.Client {
+	endpoint := os.Getenv("S3_ENDPOINT")
+	if endpoint == "" {
+		endpoint = "http://localhost:8333"
+	}
+	accessKey := os.Getenv("S3_ACCESS_KEY")
+	if accessKey == "" {
+		accessKey = "some_access_key1"
+	}
+	secretKey := os.Getenv("S3_SECRET_KEY")
+	if secretKey == "" {
+		secretKey = "some_secret_key1"
+	}
+
+	cfg, err := config.LoadDefaultConfig(
+		context.TODO(),
+		config.WithRegion("us-east-1"),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
+			accessKey,
+			secretKey,
+			"",
+		)),
+	)
+	require.NoError(t, err)
+	return s3.NewFromConfig(cfg, func(o *s3.Options) {
+		o.UsePathStyle = true
+		o.BaseEndpoint = aws.String(endpoint)
+	})
+}
+
+func createVersionedTestBucket(t *testing.T, client *s3.Client) string {
+	bucketName := "test-acl-versioned-" + strings.ToLower(strings.ReplaceAll(time.Now().Format("2006-01-02-15-04-05.000"), ":", "-"))
+	_, err := client.CreateBucket(context.TODO(), &s3.CreateBucketInput{
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err)
+
+	_, err = client.PutBucketVersioning(context.TODO(), &s3.PutBucketVersioningInput{
+		Bucket: aws.String(bucketName),
+		VersioningConfiguration: &types.VersioningConfiguration{
+			Status: types.BucketVersioningStatusEnabled,
+		},
+	})
+	require.NoError(t, err)
+	time.Sleep(100 * time.Millisecond)
+	return bucketName
+}
+
+func cleanupTestBucket(t *testing.T, client *s3.Client, bucketName string) {
+	listResp, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{
+		Bucket: aws.String(bucketName),
+	})
+	if err == nil {
+		for _, obj := range listResp.Contents {
+			client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+				Bucket: aws.String(bucketName),
+				Key:    obj.Key,
+			})
+		}
+	}
+
+	listVersionsResp, err := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{
+		Bucket: aws.String(bucketName),
+	})
+	if err == nil {
+		for _, version := range listVersionsResp.Versions {
+			client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+				Bucket:    aws.String(bucketName),
+				Key:       version.Key,
+				VersionId: version.VersionId,
+			})
+		}
+		for _, marker := range listVersionsResp.DeleteMarkers {
+			client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+				Bucket:    aws.String(bucketName),
+				Key:       marker.Key,
+				VersionId: marker.VersionId,
+			})
+		}
+	}
+
+	client.DeleteBucket(context.TODO(), &s3.DeleteBucketInput{
+		Bucket: aws.String(bucketName),
+	})
+}
+
+// TestGetObjectAclOnVersionedBucket tests retrieving ACL from versioned objects
+func TestGetObjectAclOnVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	objectKey := "versioned-object-acl"
+	putResp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Hello, ACL World!"),
+	})
+	require.NoError(t, err)
+
+	aclResp, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp.Owner)
+
+	t.Logf("Successfully retrieved ACL for versioned object %s (versionId: %s)", objectKey, *putResp.VersionId)
+}
+
+// TestGetObjectAclOnSpecificVersionInVersionedBucket tests retrieving ACL for specific versions
+func TestGetObjectAclOnSpecificVersionInVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	objectKey := "multi-version-object-acl"
+
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Version 1"),
+	})
+	require.NoError(t, err)
+
+	time.Sleep(50 * time.Millisecond)
+
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Version 2"),
+	})
+	require.NoError(t, err)
+
+	versionId1 := *version1Resp.VersionId
+	versionId2 := *version2Resp.VersionId
+
+	aclResp1, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp1.Owner)
+
+	aclResp2, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp2.Owner)
+
+	t.Logf("Successfully retrieved ACL for both versions: v1=%s, v2=%s", versionId1, versionId2)
+}
+
+// TestPutObjectAclOnVersionedBucket tests setting ACL on versioned objects
+func TestPutObjectAclOnVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	objectKey := "versioned-object-put-acl"
+	putResp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Hello, Put ACL!"),
+	})
+	require.NoError(t, err)
+
+	_, err = client.PutObjectAcl(context.TODO(), &s3.PutObjectAclInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		ACL:    types.ObjectCannedACLPublicRead,
+	})
+	require.NoError(t, err)
+
+	aclResp, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp.Owner)
+
+	t.Logf("Successfully set and verified ACL for versioned object %s (versionId: %s)", objectKey, *putResp.VersionId)
+}
+
+// TestPutObjectAclOnSpecificVersionInVersionedBucket tests setting ACL on specific versions
+func TestPutObjectAclOnSpecificVersionInVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	objectKey := "multi-version-object-put-acl"
+
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Version 1"),
+	})
+	require.NoError(t, err)
+
+	time.Sleep(50 * time.Millisecond)
+
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Version 2"),
+	})
+	require.NoError(t, err)
+
+	versionId1 := *version1Resp.VersionId
+	versionId2 := *version2Resp.VersionId
+
+	_, err = client.PutObjectAcl(context.TODO(), &s3.PutObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+		ACL:       types.ObjectCannedACLPublicRead,
+	})
+	require.NoError(t, err)
+
+	_, err = client.PutObjectAcl(context.TODO(), &s3.PutObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+		ACL:       types.ObjectCannedACLPrivate,
+	})
+	require.NoError(t, err)
+
+	aclResp1, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp1.Owner)
+
+	aclResp2, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp2.Owner)
+
+	t.Logf("Successfully set ACL on both versions: v1=%s, v2=%s", versionId1, versionId2)
+}
+
+// TestModifyAclOnDifferentVersionsIndependently tests that ACL changes on one version don't affect others
+func TestModifyAclOnDifferentVersionsIndependently(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	objectKey := "versioned-independent-acl"
+
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Version 1"),
+	})
+	require.NoError(t, err)
+
+	time.Sleep(50 * time.Millisecond)
+
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   strings.NewReader("Version 2"),
+	})
+	require.NoError(t, err)
+
+	versionId1 := *version1Resp.VersionId
+	versionId2 := *version2Resp.VersionId
+
+	_, err = client.PutObjectAcl(context.TODO(), &s3.PutObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+		ACL:       types.ObjectCannedACLPublicRead,
+	})
+	require.NoError(t, err)
+
+	_, err = client.PutObjectAcl(context.TODO(), &s3.PutObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+		ACL:       types.ObjectCannedACLPrivate,
+	})
+	require.NoError(t, err)
+
+	aclRespLatest, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclRespLatest.Owner)
+
+	aclResp1, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp1.Owner)
+
+	aclResp2, err := client.GetObjectAcl(context.TODO(), &s3.GetObjectAclInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err)
+	assert.NotNil(t, aclResp2.Owner)
+
+	t.Logf("Successfully verified independent ACL management across versions")
+}
diff --git a/test/s3/delete/s3_multi_delete_versioning_test.go b/test/s3/delete/s3_multi_delete_versioning_test.go
new file mode 100644
index 000000000..22179c611
--- /dev/null
+++ b/test/s3/delete/s3_multi_delete_versioning_test.go
@@ -0,0 +1,164 @@
+package delete
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testEndpoint  = "http://localhost:8333"
+	testAccessKey = "admin"
+	testSecretKey = "admin"
+	testRegion    = "us-east-1"
+)
+
+func getTestClient(t *testing.T) *s3.Client {
+	cfg, err := config.LoadDefaultConfig(context.TODO(),
+		config.WithRegion(testRegion),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
+			testAccessKey,
+			testSecretKey,
+			"",
+		)),
+	)
+	require.NoError(t, err)
+
+	client := s3.NewFromConfig(cfg, func(o *s3.Options) {
+		o.BaseEndpoint = aws.String(testEndpoint)
+		o.UsePathStyle = true
+	})
+
+	return client
+}
+
+func createTestBucket(t *testing.T, client *s3.Client) string {
+	bucketName := fmt.Sprintf("test-multi-delete-%d", time.Now().UnixNano())
+	_, err := client.CreateBucket(context.TODO(), &s3.CreateBucketInput{
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err)
+
+	_, err = client.PutBucketVersioning(context.TODO(), &s3.PutBucketVersioningInput{
+		Bucket: aws.String(bucketName),
+		VersioningConfiguration: &types.VersioningConfiguration{
+			Status: types.BucketVersioningStatusEnabled,
+		},
+	})
+	require.NoError(t, err)
+
+	return bucketName
+}
+
+func cleanupBucket(t *testing.T, client *s3.Client, bucket string) {
+	listResp, _ := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{
+		Bucket: aws.String(bucket),
+	})
+
+	if listResp != nil {
+		var objectsToDelete []types.ObjectIdentifier
+
+		for _, version := range listResp.Versions {
+			objectsToDelete = append(objectsToDelete, types.ObjectIdentifier{
+				Key:       version.Key,
+				VersionId: version.VersionId,
+			})
+		}
+
+		for _, marker := range listResp.DeleteMarkers {
+			objectsToDelete = append(objectsToDelete, types.ObjectIdentifier{
+				Key:       marker.Key,
+				VersionId: marker.VersionId,
+			})
+		}
+
+		if len(objectsToDelete) > 0 {
+			_, _ = client.DeleteObjects(context.TODO(), &s3.DeleteObjectsInput{
+				Bucket: aws.String(bucket),
+				Delete: &types.Delete{
+					Objects: objectsToDelete,
+					Quiet:   aws.Bool(false),
+				},
+			})
+		}
+	}
+
+	_, _ = client.DeleteBucket(context.TODO(), &s3.DeleteBucketInput{
+		Bucket: aws.String(bucket),
+	})
+}
+
+func TestVersioningMultiObjectDelete(t *testing.T) {
+	client := getTestClient(t)
+	bucket := createTestBucket(t, client)
+	defer cleanupBucket(t, client, bucket)
+
+	key := "key"
+	numVersions := 2
+	var versionIds []string
+
+	for i := 0; i < numVersions; i++ {
+		content := fmt.Sprintf("content-%d", i)
+		putResp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+			Body:   bytes.NewReader([]byte(content)),
+		})
+		require.NoError(t, err)
+		require.NotNil(t, putResp.VersionId)
+		versionIds = append(versionIds, *putResp.VersionId)
+	}
+
+	assert.Len(t, versionIds, 2)
+
+	var objectsToDelete []types.ObjectIdentifier
+	for _, vid := range versionIds {
+		objectsToDelete = append(objectsToDelete, types.ObjectIdentifier{
+			Key:       aws.String(key),
+			VersionId: aws.String(vid),
+		})
+	}
+
+	deleteResp, err := client.DeleteObjects(context.TODO(), &s3.DeleteObjectsInput{
+		Bucket: aws.String(bucket),
+		Delete: &types.Delete{
+			Objects: objectsToDelete,
+		},
+	})
+	require.NoError(t, err)
+	t.Logf("Delete response: Deleted=%d, Errors=%d", len(deleteResp.Deleted), len(deleteResp.Errors))
+
+	listResp, err := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{
+		Bucket: aws.String(bucket),
+	})
+	require.NoError(t, err)
+
+	if listResp.Versions != nil && len(listResp.Versions) > 0 {
+		t.Errorf("FAIL: Expected no versions, but found %d versions:", len(listResp.Versions))
+		for _, v := range listResp.Versions {
+			t.Logf("  - Key=%s, VersionId=%s, IsLatest=%v", *v.Key, *v.VersionId, v.IsLatest)
+		}
+	} else {
+		t.Logf("PASS: Versions correctly deleted")
+	}
+	assert.Nil(t, listResp.Versions, "Expected no versions after deletion")
+
+	deleteResp2, err := client.DeleteObjects(context.TODO(), &s3.DeleteObjectsInput{
+		Bucket: aws.String(bucket),
+		Delete: &types.Delete{
+			Objects: objectsToDelete,
+		},
+	})
+	require.NoError(t, err)
+	assert.Empty(t, deleteResp2.Errors, "Idempotent delete should not return errors")
+}
diff --git a/test/s3/tagging/s3_tagging_test.go b/test/s3/tagging/s3_tagging_test.go
index c490ca1aa..4606ec800 100644
--- a/test/s3/tagging/s3_tagging_test.go
+++ b/test/s3/tagging/s3_tagging_test.go
@@ -63,18 +63,12 @@ func getS3Client(t *testing.T) *s3.Client {
 			defaultConfig.SecretKey,
 			"",
 		)),
-		config.WithEndpointResolverWithOptions(aws.EndpointResolverWithOptionsFunc(
-			func(service, region string, options ...interface{}) (aws.Endpoint, error) {
-				return aws.Endpoint{
-					URL:           defaultConfig.Endpoint,
-					SigningRegion: defaultConfig.Region,
-				}, nil
-			})),
 	)
 	require.NoError(t, err)
 
 	client := s3.NewFromConfig(cfg, func(o *s3.Options) {
 		o.UsePathStyle = true
+		o.BaseEndpoint = aws.String(defaultConfig.Endpoint)
 	})
 	return client
 }
@@ -113,6 +107,33 @@ func cleanupTestBucket(t *testing.T, client *s3.Client, bucketName string) {
 		}
 	}
 
+	// Delete all versions and delete markers if versioning is enabled
+	listVersionsResp, err := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{
+		Bucket: aws.String(bucketName),
+	})
+	if err == nil {
+		for _, version := range listVersionsResp.Versions {
+			_, err := client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+				Bucket:    aws.String(bucketName),
+				Key:       version.Key,
+				VersionId: version.VersionId,
+			})
+			if err != nil {
+				t.Logf("Warning: failed to delete version %s: %v", *version.Key, err)
+			}
+		}
+		for _, marker := range listVersionsResp.DeleteMarkers {
+			_, err := client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+				Bucket:    aws.String(bucketName),
+				Key:       marker.Key,
+				VersionId: marker.VersionId,
+			})
+			if err != nil {
+				t.Logf("Warning: failed to delete marker %s: %v", *marker.Key, err)
+			}
+		}
+	}
+
 	// Then delete the bucket
 	_, err = client.DeleteBucket(context.TODO(), &s3.DeleteBucketInput{
 		Bucket: aws.String(bucketName),
diff --git a/test/s3/tagging/s3_tagging_versioning_test.go b/test/s3/tagging/s3_tagging_versioning_test.go
new file mode 100644
index 000000000..5aa49f956
--- /dev/null
+++ b/test/s3/tagging/s3_tagging_versioning_test.go
@@ -0,0 +1,434 @@
+package tagging
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// This is the fix for GitHub issue #7868 where tagging failed with "no entry is found in filer store"
+// TestPutObjectTaggingOnVersionedBucket tests setting tags on objects in a versioned bucket
+func TestPutObjectTaggingOnVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	// Put object in versioned bucket
+	objectKey := "versioned-object-with-tags"
+	objectContent := "Hello, Versioned World!"
+	_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:   strings.NewReader(objectContent),
+		Key:    aws.String(objectKey),
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err, "Should be able to put object in versioned bucket")
+
+	// Set tags on the object in versioned bucket
+	_, err = client.PutObjectTagging(context.TODO(), &s3.PutObjectTaggingInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{
+				{
+					Key:   aws.String("env"),
+					Value: aws.String("production"),
+				},
+				{
+					Key:   aws.String("team"),
+					Value: aws.String("platform"),
+				},
+			},
+		},
+	})
+	require.NoError(t, err, "Should be able to put tags on versioned object")
+
+	// Get the tags back
+	tagResp, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err, "Should be able to get tags from versioned object")
+
+	// Verify tags
+	assert.Len(t, tagResp.TagSet, 2, "Should have 2 tags")
+	tagMap := make(map[string]string)
+	for _, tag := range tagResp.TagSet {
+		tagMap[*tag.Key] = *tag.Value
+	}
+	assert.Equal(t, "production", tagMap["env"], "env tag should be 'production'")
+	assert.Equal(t, "platform", tagMap["team"], "team tag should be 'platform'")
+}
+
+// TestPutObjectTaggingOnSpecificVersionInVersionedBucket tests setting tags on a specific version
+func TestPutObjectTaggingOnSpecificVersionInVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	// Create multiple versions of the object
+	objectKey := "multi-version-object"
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:   strings.NewReader("Version 1"),
+		Key:    aws.String(objectKey),
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err)
+
+	// Small delay to ensure different version IDs
+	time.Sleep(50 * time.Millisecond)
+	versionId1 := *version1Resp.VersionId
+
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:   strings.NewReader("Version 2"),
+		Key:    aws.String(objectKey),
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err)
+	versionId2 := *version2Resp.VersionId
+
+	// Set tags on version 1
+	_, err = client.PutObjectTagging(context.TODO(), &s3.PutObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{
+				{
+					Key:   aws.String("version"),
+					Value: aws.String("v1"),
+				},
+			},
+		},
+	})
+	require.NoError(t, err, "Should be able to put tags on specific version 1")
+
+	// Set tags on version 2
+	_, err = client.PutObjectTagging(context.TODO(), &s3.PutObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{
+				{
+					Key:   aws.String("version"),
+					Value: aws.String("v2"),
+				},
+			},
+		},
+	})
+	require.NoError(t, err, "Should be able to put tags on specific version 2")
+
+	// Get tags from version 1
+	tagResp1, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err, "Should be able to get tags from version 1")
+	assert.Len(t, tagResp1.TagSet, 1, "Version 1 should have 1 tag")
+	assert.Equal(t, "v1", *tagResp1.TagSet[0].Value, "Version 1 tag value should be 'v1'")
+
+	// Get tags from version 2
+	tagResp2, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err, "Should be able to get tags from version 2")
+	assert.Len(t, tagResp2.TagSet, 1, "Version 2 should have 1 tag")
+	assert.Equal(t, "v2", *tagResp2.TagSet[0].Value, "Version 2 tag value should be 'v2'")
+}
+
+// TestDeleteObjectTaggingOnVersionedBucket tests deleting tags from versioned objects
+func TestDeleteObjectTaggingOnVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	// Put object with tags in versioned bucket
+	objectKey := "versioned-object-tag-delete"
+	objectContent := "Hello, Delete Tags!"
+	_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:    strings.NewReader(objectContent),
+		Key:     aws.String(objectKey),
+		Bucket:  aws.String(bucketName),
+		Tagging: aws.String("env=dev&purpose=testing"),
+	})
+	require.NoError(t, err)
+
+	// Verify tags exist
+	tagResp, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err)
+	assert.Len(t, tagResp.TagSet, 2, "Should have 2 tags before deletion")
+
+	// Delete tags from the versioned object
+	_, err = client.DeleteObjectTagging(context.TODO(), &s3.DeleteObjectTaggingInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err, "Should be able to delete tags from versioned object")
+
+	// Verify tags are deleted
+	tagResp, err = client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err)
+	assert.Len(t, tagResp.TagSet, 0, "Should have 0 tags after deletion")
+}
+
+// TestDeleteObjectTaggingOnSpecificVersionInVersionedBucket tests deleting tags on a specific version
+func TestDeleteObjectTaggingOnSpecificVersionInVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	// Create two versions with tags
+	objectKey := "versioned-multi-delete-tags"
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:    strings.NewReader("Version 1"),
+		Key:     aws.String(objectKey),
+		Bucket:  aws.String(bucketName),
+		Tagging: aws.String("version=v1&keep=true"),
+	})
+	require.NoError(t, err)
+
+	// Small delay to ensure different version IDs
+	time.Sleep(50 * time.Millisecond)
+	versionId1 := *version1Resp.VersionId
+
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:    strings.NewReader("Version 2"),
+		Key:     aws.String(objectKey),
+		Bucket:  aws.String(bucketName),
+		Tagging: aws.String("version=v2&keep=true"),
+	})
+	require.NoError(t, err)
+	versionId2 := *version2Resp.VersionId
+
+	// Delete tags from version 1 only
+	_, err = client.DeleteObjectTagging(context.TODO(), &s3.DeleteObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err, "Should be able to delete tags from version 1")
+
+	// Verify version 1 has no tags
+	tagResp1, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err)
+	assert.Len(t, tagResp1.TagSet, 0, "Version 1 should have 0 tags after deletion")
+
+	// Verify version 2 still has tags
+	tagResp2, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err)
+	assert.Len(t, tagResp2.TagSet, 2, "Version 2 should still have 2 tags")
+}
+
+// TestGetObjectTaggingOnVersionedBucket tests retrieving tags from versioned objects
+func TestGetObjectTaggingOnVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	// Create multiple versions
+	objectKey := "versioned-object-get-tags"
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:    strings.NewReader("Version 1"),
+		Key:     aws.String(objectKey),
+		Bucket:  aws.String(bucketName),
+		Tagging: aws.String("v=1&stage=dev"),
+	})
+	require.NoError(t, err)
+
+	versionId1 := *version1Resp.VersionId
+	time.Sleep(50 * time.Millisecond)
+
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:    strings.NewReader("Version 2"),
+		Key:     aws.String(objectKey),
+		Bucket:  aws.String(bucketName),
+		Tagging: aws.String("v=2&stage=prod"),
+	})
+	require.NoError(t, err)
+
+	versionId2 := *version2Resp.VersionId
+
+	// Get tags from specific versions
+	tagResp1, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err, "Should be able to get tags from version 1")
+	assert.Len(t, tagResp1.TagSet, 2, "Version 1 should have 2 tags")
+	tagMap1 := make(map[string]string)
+	for _, tag := range tagResp1.TagSet {
+		tagMap1[*tag.Key] = *tag.Value
+	}
+	assert.Equal(t, "1", tagMap1["v"], "Version 1 should have v=1")
+	assert.Equal(t, "dev", tagMap1["stage"], "Version 1 should have stage=dev")
+
+	tagResp2, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err, "Should be able to get tags from version 2")
+	assert.Len(t, tagResp2.TagSet, 2, "Version 2 should have 2 tags")
+	tagMap2 := make(map[string]string)
+	for _, tag := range tagResp2.TagSet {
+		tagMap2[*tag.Key] = *tag.Value
+	}
+	assert.Equal(t, "2", tagMap2["v"], "Version 2 should have v=2")
+	assert.Equal(t, "prod", tagMap2["stage"], "Version 2 should have stage=prod")
+
+	// Get tags from latest version (should be version 2)
+	tagRespLatest, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	require.NoError(t, err, "Should be able to get tags from latest version")
+	assert.Len(t, tagRespLatest.TagSet, 2, "Latest version should have 2 tags")
+	tagMapLatest := make(map[string]string)
+	for _, tag := range tagRespLatest.TagSet {
+		tagMapLatest[*tag.Key] = *tag.Value
+	}
+	assert.Equal(t, "2", tagMapLatest["v"], "Latest version should have v=2")
+	assert.Equal(t, "prod", tagMapLatest["stage"], "Latest version should have stage=prod")
+}
+
+// TestModifyTagsOnVersionedObject tests changing tags on different versions independently
+func TestModifyTagsOnVersionedObject(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := createVersionedTestBucket(t, client)
+	defer cleanupTestBucket(t, client, bucketName)
+
+	// Create version 1
+	objectKey := "versioned-modify-tags"
+	version1Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:   strings.NewReader("Version 1"),
+		Key:    aws.String(objectKey),
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err)
+
+	versionId1 := *version1Resp.VersionId
+	time.Sleep(50 * time.Millisecond)
+
+	// Create version 2
+	version2Resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Body:   strings.NewReader("Version 2"),
+		Key:    aws.String(objectKey),
+		Bucket: aws.String(bucketName),
+	})
+	require.NoError(t, err)
+
+	versionId2 := *version2Resp.VersionId
+
+	// Add tags to version 1
+	_, err = client.PutObjectTagging(context.TODO(), &s3.PutObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{
+				{
+					Key:   aws.String("status"),
+					Value: aws.String("old"),
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	// Modify tags on version 1 (replace status and add new tag)
+	_, err = client.PutObjectTagging(context.TODO(), &s3.PutObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{
+				{
+					Key:   aws.String("status"),
+					Value: aws.String("archived"),
+				},
+				{
+					Key:   aws.String("archived-date"),
+					Value: aws.String("2024-01-01"),
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	// Add tags to version 2
+	_, err = client.PutObjectTagging(context.TODO(), &s3.PutObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{
+				{
+					Key:   aws.String("status"),
+					Value: aws.String("current"),
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	// Verify final state
+	tagResp1, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId1),
+	})
+	require.NoError(t, err)
+	assert.Len(t, tagResp1.TagSet, 2, "Version 1 should have 2 tags after modification")
+
+	tagResp2, err := client.GetObjectTagging(context.TODO(), &s3.GetObjectTaggingInput{
+		Bucket:    aws.String(bucketName),
+		Key:       aws.String(objectKey),
+		VersionId: aws.String(versionId2),
+	})
+	require.NoError(t, err)
+	assert.Len(t, tagResp2.TagSet, 1, "Version 2 should have 1 tag")
+}
+
+// createVersionedTestBucket creates a test bucket with versioning enabled
+func createVersionedTestBucket(t *testing.T, client *s3.Client) string {
+	bucketName := createTestBucket(t, client)
+
+	// Enable versioning on the bucket
+	_, err := client.PutBucketVersioning(context.TODO(), &s3.PutBucketVersioningInput{
+		Bucket: aws.String(bucketName),
+		VersioningConfiguration: &types.VersioningConfiguration{
+			Status: types.BucketVersioningStatusEnabled,
+		},
+	})
+	require.NoError(t, err, "Should be able to enable versioning")
+
+	// Wait for versioning configuration to be applied
+	time.Sleep(100 * time.Millisecond)
+
+	return bucketName
+}
diff --git a/weed/s3api/s3api_object_handlers_acl.go b/weed/s3api/s3api_object_handlers_acl.go
index e90d84603..212354a30 100644
--- a/weed/s3api/s3api_object_handlers_acl.go
+++ b/weed/s3api/s3api_object_handlers_acl.go
@@ -306,7 +306,7 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
 	if versioningConfigured {
 		if versionId != "" && versionId != "null" {
 			// Versioned object - update the specific version file in .versions directory
-			updateDirectory = s3a.option.BucketsPath + "/" + bucket + "/" + object + s3_constants.VersionsFolder
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket + object + s3_constants.VersionsFolder
 		} else {
 			// Latest version in versioned bucket - could be null version or versioned object
 			// Extract version ID from the entry to determine where it's stored
@@ -322,7 +322,7 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
 				updateDirectory = s3a.option.BucketsPath + "/" + bucket
 			} else {
 				// Versioned object - stored in .versions directory
-				updateDirectory = s3a.option.BucketsPath + "/" + bucket + "/" + object + s3_constants.VersionsFolder
+				updateDirectory = s3a.option.BucketsPath + "/" + bucket + object + s3_constants.VersionsFolder
 			}
 		}
 	} else {
diff --git a/weed/s3api/s3api_object_handlers_tagging.go b/weed/s3api/s3api_object_handlers_tagging.go
index 23ca05133..7b6b947da 100644
--- a/weed/s3api/s3api_object_handlers_tagging.go
+++ b/weed/s3api/s3api_object_handlers_tagging.go
@@ -1,12 +1,14 @@
 package s3api
 
 import (
+	"context"
 	"encoding/xml"
 	"fmt"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"io"
 	"net/http"
 
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
@@ -20,21 +22,79 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R
 	bucket, object := s3_constants.GetBucketAndObject(r)
 	glog.V(3).Infof("GetObjectTaggingHandler %s %s", bucket, object)
 
-	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
-	dir, name := target.DirAndName()
+	// Check for specific version ID in query parameters
+	versionId := r.URL.Query().Get("versionId")
 
-	tags, err := s3a.getTags(dir, name)
+	// Check if versioning is configured for the bucket (Enabled or Suspended)
+	versioningConfigured, err := s3a.isVersioningConfigured(bucket)
 	if err != nil {
 		if err == filer_pb.ErrNotFound {
-			glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
+			return
+		}
+		glog.Errorf("GetObjectTaggingHandler: Error checking versioning status for bucket %s: %v", bucket, err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		return
+	}
+
+	var entry *filer_pb.Entry
+
+	if versioningConfigured {
+		// Handle versioned object tagging retrieval
+		if versionId != "" {
+			// Request for specific version
+			glog.V(2).Infof("GetObjectTaggingHandler: requesting tags for specific version %s of %s%s", versionId, bucket, object)
+			entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId)
 		} else {
-			glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			// Request for latest version
+			glog.V(2).Infof("GetObjectTaggingHandler: requesting tags for latest version of %s%s", bucket, object)
+			entry, err = s3a.getLatestObjectVersion(bucket, object)
+		}
+
+		if err != nil {
+			glog.Errorf("GetObjectTaggingHandler: Failed to get object version %s for %s%s: %v", versionId, bucket, object, err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			return
+		}
+
+		// Check if this is a delete marker
+		if entry.Extended != nil {
+			if deleteMarker, exists := entry.Extended[s3_constants.ExtDeleteMarkerKey]; exists && string(deleteMarker) == "true" {
+				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+				return
+			}
 		}
+	} else {
+		// Handle regular (non-versioned) object tagging retrieval
+		target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
+		dir, name := target.DirAndName()
+
+		tags, err := s3a.getTags(dir, name)
+		if err != nil {
+			if err == filer_pb.ErrNotFound {
+				glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			} else {
+				glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			}
+			return
+		}
+
+		writeSuccessResponseXML(w, r, FromTags(tags))
 		return
 	}
 
+	// Extract tags from the entry's extended attributes
+	tags := make(map[string]string)
+	if entry.Extended != nil {
+		for k, v := range entry.Extended {
+			if len(k) > len(S3TAG_PREFIX) && k[:len(S3TAG_PREFIX)] == S3TAG_PREFIX {
+				tags[k[len(S3TAG_PREFIX):]] = string(v)
+			}
+		}
+	}
+
 	writeSuccessResponseXML(w, r, FromTags(tags))
 
 }
@@ -46,9 +106,6 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 	bucket, object := s3_constants.GetBucketAndObject(r)
 	glog.V(3).Infof("PutObjectTaggingHandler %s %s", bucket, object)
 
-	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
-	dir, name := target.DirAndName()
-
 	tagging := &Tagging{}
 	input, err := io.ReadAll(io.LimitReader(r.Body, r.ContentLength))
 	if err != nil {
@@ -69,17 +126,133 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 		return
 	}
 
-	if err = s3a.setTags(dir, name, tagging.ToTags()); err != nil {
+	// Check for specific version ID in query parameters
+	versionId := r.URL.Query().Get("versionId")
+
+	// Check if versioning is configured for the bucket (Enabled or Suspended)
+	versioningConfigured, err := s3a.isVersioningConfigured(bucket)
+	if err != nil {
 		if err == filer_pb.ErrNotFound {
-			glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
+			return
+		}
+		glog.Errorf("PutObjectTaggingHandler: Error checking versioning status for bucket %s: %v", bucket, err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		return
+	}
+
+	var entry *filer_pb.Entry
+
+	if versioningConfigured {
+		// Handle versioned object tagging modification
+		if versionId != "" {
+			// Request for specific version
+			glog.V(2).Infof("PutObjectTaggingHandler: modifying tags for specific version %s of %s%s", versionId, bucket, object)
+			entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId)
+		} else {
+			// Request for latest version
+			glog.V(2).Infof("PutObjectTaggingHandler: modifying tags for latest version of %s%s", bucket, object)
+			entry, err = s3a.getLatestObjectVersion(bucket, object)
+		}
+
+		if err != nil {
+			glog.Errorf("PutObjectTaggingHandler: Failed to get object version %s for %s%s: %v", versionId, bucket, object, err)
 			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			return
+		}
+
+		// Check if this is a delete marker
+		if entry.Extended != nil {
+			if deleteMarker, exists := entry.Extended[s3_constants.ExtDeleteMarkerKey]; exists && string(deleteMarker) == "true" {
+				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+				return
+			}
+		}
+	} else {
+		// Handle regular (non-versioned) object tagging modification
+		target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
+		dir, name := target.DirAndName()
+
+		if err = s3a.setTags(dir, name, tags); err != nil {
+			if err == filer_pb.ErrNotFound {
+				glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			} else {
+				glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			}
+			return
+		}
+
+		w.WriteHeader(http.StatusOK)
+		s3err.PostLog(r, http.StatusOK, s3err.ErrNone)
+		return
+	}
+
+	// For versioned objects, determine the correct directory based on the version
+	var updateDirectory string
+	if versionId != "" {
+		// Specific version requested
+		if versionId == "null" {
+			// Null version (pre-versioning object) - stored as regular file
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket
 		} else {
-			glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			// Versioned object - stored in .versions directory
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket + object + s3_constants.VersionsFolder
+		}
+	} else {
+		// Latest version in versioned bucket - could be null version or versioned object
+		// Extract version ID from the entry to determine where it's stored
+		var actualVersionId string
+		if entry.Extended != nil {
+			if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists {
+				actualVersionId = string(versionIdBytes)
+			}
+		}
+
+		if actualVersionId == "null" || actualVersionId == "" {
+			// Null version (pre-versioning object) - stored as regular file
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket
+		} else {
+			// Versioned object - stored in .versions directory
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket + object + s3_constants.VersionsFolder
+		}
+	}
+
+	// Remove old tags and add new ones
+	for k := range entry.Extended {
+		if len(k) > len(S3TAG_PREFIX) && k[:len(S3TAG_PREFIX)] == S3TAG_PREFIX {
+			delete(entry.Extended, k)
+		}
+	}
+
+	if entry.Extended == nil {
+		entry.Extended = make(map[string][]byte)
+	}
+	for k, v := range tags {
+		entry.Extended[S3TAG_PREFIX+k] = []byte(v)
+	}
+
+	// Update the entry with tags
+	err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		request := &filer_pb.UpdateEntryRequest{
+			Directory: updateDirectory,
+			Entry:     entry,
+		}
+
+		if _, err := client.UpdateEntry(context.Background(), request); err != nil {
+			return err
 		}
+		return nil
+	})
+
+	if err != nil {
+		glog.Errorf("PutObjectTaggingHandler: failed to update entry: %v", err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
+	glog.V(3).Infof("PutObjectTaggingHandler: Successfully updated tags for %s/%s", bucket, object)
 	w.WriteHeader(http.StatusOK)
 	s3err.PostLog(r, http.StatusOK, s3err.ErrNone)
 }
@@ -91,21 +264,136 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt
 	bucket, object := s3_constants.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteObjectTaggingHandler %s %s", bucket, object)
 
-	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
-	dir, name := target.DirAndName()
+	// Check for specific version ID in query parameters
+	versionId := r.URL.Query().Get("versionId")
 
-	err := s3a.rmTags(dir, name)
+	// Check if versioning is configured for the bucket (Enabled or Suspended)
+	versioningConfigured, err := s3a.isVersioningConfigured(bucket)
 	if err != nil {
 		if err == filer_pb.ErrNotFound {
-			glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
+			return
+		}
+		glog.Errorf("DeleteObjectTaggingHandler: Error checking versioning status for bucket %s: %v", bucket, err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		return
+	}
+
+	var entry *filer_pb.Entry
+
+	if versioningConfigured {
+		// Handle versioned object tagging deletion
+		if versionId != "" {
+			// Request for specific version
+			glog.V(2).Infof("DeleteObjectTaggingHandler: deleting tags for specific version %s of %s%s", versionId, bucket, object)
+			entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId)
+		} else {
+			// Request for latest version
+			glog.V(2).Infof("DeleteObjectTaggingHandler: deleting tags for latest version of %s%s", bucket, object)
+			entry, err = s3a.getLatestObjectVersion(bucket, object)
+		}
+
+		if err != nil {
+			glog.Errorf("DeleteObjectTaggingHandler: Failed to get object version %s for %s%s: %v", versionId, bucket, object, err)
 			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			return
+		}
+
+		// Check if this is a delete marker
+		if entry.Extended != nil {
+			if deleteMarker, exists := entry.Extended[s3_constants.ExtDeleteMarkerKey]; exists && string(deleteMarker) == "true" {
+				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+				return
+			}
+		}
+	} else {
+		// Handle regular (non-versioned) object tagging deletion
+		target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
+		dir, name := target.DirAndName()
+
+		err := s3a.rmTags(dir, name)
+		if err != nil {
+			if err == filer_pb.ErrNotFound {
+				glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+			} else {
+				glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			}
+			return
+		}
+
+		w.WriteHeader(http.StatusNoContent)
+		s3err.PostLog(r, http.StatusNoContent, s3err.ErrNone)
+		return
+	}
+
+	// For versioned objects, determine the correct directory based on the version
+	var updateDirectory string
+	if versionId != "" {
+		// Specific version requested
+		if versionId == "null" {
+			// Null version (pre-versioning object) - stored as regular file
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket
 		} else {
-			glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			// Versioned object - stored in .versions directory
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket + object + s3_constants.VersionsFolder
+		}
+	} else {
+		// Latest version in versioned bucket - could be null version or versioned object
+		// Extract version ID from the entry to determine where it's stored
+		var actualVersionId string
+		if entry.Extended != nil {
+			if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists {
+				actualVersionId = string(versionIdBytes)
+			}
+		}
+
+		if actualVersionId == "null" || actualVersionId == "" {
+			// Null version (pre-versioning object) - stored as regular file
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket
+		} else {
+			// Versioned object - stored in .versions directory
+			updateDirectory = s3a.option.BucketsPath + "/" + bucket + object + s3_constants.VersionsFolder
+		}
+	}
+
+	// Remove all tags
+	hasDeletion := false
+	for k := range entry.Extended {
+		if len(k) > len(S3TAG_PREFIX) && k[:len(S3TAG_PREFIX)] == S3TAG_PREFIX {
+			delete(entry.Extended, k)
+			hasDeletion = true
+		}
+	}
+
+	if !hasDeletion {
+		// No tags to delete - success
+		w.WriteHeader(http.StatusNoContent)
+		s3err.PostLog(r, http.StatusNoContent, s3err.ErrNone)
+		return
+	}
+
+	// Update the entry
+	err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		request := &filer_pb.UpdateEntryRequest{
+			Directory: updateDirectory,
+			Entry:     entry,
+		}
+
+		if _, err := client.UpdateEntry(context.Background(), request); err != nil {
+			return err
 		}
+		return nil
+	})
+
+	if err != nil {
+		glog.Errorf("DeleteObjectTaggingHandler: failed to update entry: %v", err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
+	glog.V(3).Infof("DeleteObjectTaggingHandler: Successfully deleted tags for %s/%s", bucket, object)
 	w.WriteHeader(http.StatusNoContent)
 	s3err.PostLog(r, http.StatusNoContent, s3err.ErrNone)
 }
diff --git a/weed/s3api/s3api_version_id.go b/weed/s3api/s3api_version_id.go
index 0ea3e6f89..db347d927 100644
--- a/weed/s3api/s3api_version_id.go
+++ b/weed/s3api/s3api_version_id.go
@@ -184,4 +184,3 @@ func (s3a *S3ApiServer) generateVersionIdForObject(bucket, object string) string
 	useInvertedFormat := s3a.getVersionIdFormat(bucket, object)
 	return generateVersionId(useInvertedFormat)
 }
-