diff --git a/.github/workflows/s3-example-integration-tests.yml b/.github/workflows/s3-example-integration-tests.yml
index 23e296773..4e39c3bc0 100644
--- a/.github/workflows/s3-example-integration-tests.yml
+++ b/.github/workflows/s3-example-integration-tests.yml
@@ -64,6 +64,14 @@ jobs:
           echo "=== Running S3 Empty Directory Marker Tests ==="
           go test -v -timeout=180s -run TestS3ListObjectsEmptyDirectoryMarkers ./...
 
+      - name: Run S3 Prefix Object Tests
+        timeout-minutes: 15
+        working-directory: test/s3/normal
+        run: |
+          set -x
+          echo "=== Running S3 Prefix Object Tests ==="
+          go test -v -timeout=180s -run TestS3PrefixObjectKeys ./...
+
       - name: Run IAM Integration Tests
         timeout-minutes: 15
         working-directory: test/s3/normal
diff --git a/test/s3/copying/s3_rename_test.go b/test/s3/copying/s3_rename_test.go
index 881f4cca0..a3dc62e92 100644
--- a/test/s3/copying/s3_rename_test.go
+++ b/test/s3/copying/s3_rename_test.go
@@ -156,8 +156,9 @@ func TestRenameObjectSourceIfMatch(t *testing.T) {
 	assert.True(t, objectExists(t, client, bucketName, "target.txt"))
 }
 
-// TestRenameObjectOntoDirectory: a key that already holds other objects is a
-// directory, and an object must not be allowed to replace one.
+// TestRenameObjectOntoDirectory: S3 keys are flat, so a key that other keys are
+// nested under is still a key of its own. The rename writes it without disturbing
+// them - it does not replace the directory, it stores the object on it.
 func TestRenameObjectOntoDirectory(t *testing.T) {
 	client := getS3Client(t)
 	bucketName := getNewBucketName()
@@ -172,9 +173,10 @@ func TestRenameObjectOntoDirectory(t *testing.T) {
 		Key:          aws.String("target"),
 		RenameSource: aws.String(createRenameSource("source.txt")),
 	})
-	requireRenameStatus(t, err, 409)
-	assert.True(t, objectExists(t, client, bucketName, "source.txt"))
-	assert.True(t, objectExists(t, client, bucketName, "target/child.txt"))
+	require.NoError(t, err)
+	assert.False(t, objectExists(t, client, bucketName, "source.txt"))
+	assert.Equal(t, "content", getObjectBody(t, getObject(t, client, bucketName, "target")))
+	assert.Equal(t, "child", getObjectBody(t, getObject(t, client, bucketName, "target/child.txt")))
 }
 
 // TestRenameObjectDirectorySource: a directory can be named without a trailing
diff --git a/test/s3/normal/s3_prefix_object_test.go b/test/s3/normal/s3_prefix_object_test.go
new file mode 100644
index 000000000..89fe154f0
--- /dev/null
+++ b/test/s3/normal/s3_prefix_object_test.go
@@ -0,0 +1,546 @@
+package example
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	v1credentials "github.com/aws/aws-sdk-go/aws/credentials"
+	v1signer "github.com/aws/aws-sdk-go/aws/signer/v4"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestS3PrefixObjectKeys covers keys that are a strict prefix of other keys: S3's
+// namespace is flat, so "collision/foo" and "collision/foo/bar" are independent
+// objects that coexist in either write order.
+func TestS3PrefixObjectKeys(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	cluster, err := startMiniCluster(t)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	put := func(t *testing.T, bucket, key string, body []byte) {
+		t.Helper()
+		_, err := cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+			Body:   bytes.NewReader(body),
+		})
+		require.NoError(t, err, "put %s", key)
+	}
+	// read checks both paths a client reaches an object by, since a directory entry
+	// carrying an object is served by neither the directory nor the plain object path
+	// alone.
+	read := func(t *testing.T, bucket, key string, want []byte) {
+		t.Helper()
+		head, err := cluster.s3Client.HeadObject(&s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		require.NoError(t, err, "head %s", key)
+		assert.Equal(t, int64(len(want)), aws.Int64Value(head.ContentLength), "head %s", key)
+
+		get, err := cluster.s3Client.GetObject(&s3.GetObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		require.NoError(t, err, "get %s", key)
+		defer get.Body.Close()
+		got, err := io.ReadAll(get.Body)
+		require.NoError(t, err, "read %s", key)
+		assert.Equal(t, want, got, "get %s", key)
+	}
+	// gone checks the key answers as absent rather than lingering on a directory
+	// entry that outlived the object.
+	gone := func(t *testing.T, bucket, key string) {
+		t.Helper()
+		_, err := cluster.s3Client.GetObject(&s3.GetObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		var missing awserr.RequestFailure
+		require.ErrorAs(t, err, &missing, "get %s", key)
+		assert.Equal(t, http.StatusNotFound, missing.StatusCode(), "get %s", key)
+
+		_, err = cluster.s3Client.HeadObject(&s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		require.ErrorAs(t, err, &missing, "head %s", key)
+		assert.Equal(t, http.StatusNotFound, missing.StatusCode(), "head %s", key)
+	}
+	listKeys := func(t *testing.T, bucket string) []string {
+		t.Helper()
+		resp, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{Bucket: aws.String(bucket)})
+		require.NoError(t, err)
+		keys := collectKeys(resp.Contents)
+		sort.Strings(keys)
+		return keys
+	}
+
+	body := []byte("prefix object")
+	// Distinct bodies, so a read that resolves to the wrong entry cannot pass.
+	nested := []byte("nested under the prefix object")
+
+	// The reported order: the nested key is written first, so the prefix key has to
+	// land on a path the filer already holds a directory at.
+	t.Run("ChildFirst", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-child-first-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+
+		assert.Equal(t, []string{"collision/foo", "collision/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo", body)
+		read(t, bucket, "collision/foo/bar", nested)
+	})
+
+	// The opposite order used to keep the prefix key's data but hide the key.
+	t.Run("PrefixFirst", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-first-")
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/foo/bar", nested)
+
+		assert.Equal(t, []string{"collision/foo", "collision/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo", body)
+		read(t, bucket, "collision/foo/bar", nested)
+	})
+
+	// An empty object leaves no chunks, content or mime behind, so it is the case a
+	// promoted directory carries no other trace of.
+	t.Run("EmptyObject", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-empty-")
+		put(t, bucket, "a/foo/bar", nil)
+		put(t, bucket, "a/foo", nil)
+		put(t, bucket, "b/foo", nil)
+		put(t, bucket, "b/foo/bar", nil)
+
+		assert.Equal(t, []string{"a/foo", "a/foo/bar", "b/foo", "b/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "a/foo", []byte{})
+		read(t, bucket, "b/foo", []byte{})
+	})
+
+	// The key has no trailing slash, and the keys nested under it still roll up into
+	// their own CommonPrefix.
+	t.Run("Delimiter", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-delimiter-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/other", body)
+
+		resp, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+			Bucket:    aws.String(bucket),
+			Prefix:    aws.String("collision/"),
+			Delimiter: aws.String("/"),
+		})
+		require.NoError(t, err)
+
+		keys := collectKeys(resp.Contents)
+		sort.Strings(keys)
+		assert.Equal(t, []string{"collision/foo", "collision/other"}, keys)
+		assert.Equal(t, []string{"collision/foo/"}, collectPrefixes(resp.CommonPrefixes))
+
+		// Listing the prefix itself names only what is under it.
+		resp, err = cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+			Bucket:    aws.String(bucket),
+			Prefix:    aws.String("collision/foo/"),
+			Delimiter: aws.String("/"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []string{"collision/foo/bar"}, collectKeys(resp.Contents))
+		assert.Empty(t, collectPrefixes(resp.CommonPrefixes))
+	})
+
+	// The key and the CommonPrefix its nested keys fold into come off one filer
+	// entry, so a page boundary must not drop either of them.
+	t.Run("Paged", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-paged-")
+		for _, key := range []string{"foo", "foo/bar", "foobar", "other", "zed", "zed/a"} {
+			put(t, bucket, key, body)
+		}
+
+		for _, maxKeys := range []int64{1, 2, 3, 4, 5} {
+			var keys, prefixes []string
+			var token *string
+			for page := 0; page < 12; page++ {
+				resp, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+					Bucket:            aws.String(bucket),
+					Delimiter:         aws.String("/"),
+					MaxKeys:           aws.Int64(maxKeys),
+					ContinuationToken: token,
+				})
+				require.NoError(t, err)
+				// One entry over the budget is the documented same-entry exception;
+				// anything more means the unsigned budget wrapped.
+				assert.LessOrEqual(t, int64(len(resp.Contents)+len(resp.CommonPrefixes)), maxKeys+1,
+					"maxKeys=%d page %d", maxKeys, page)
+				keys = append(keys, collectKeys(resp.Contents)...)
+				prefixes = append(prefixes, collectPrefixes(resp.CommonPrefixes)...)
+				if !aws.BoolValue(resp.IsTruncated) {
+					token = nil
+					break
+				}
+				token = resp.NextContinuationToken
+				require.NotNil(t, token, "a truncated page must name where to resume")
+			}
+			require.Nil(t, token, "maxKeys=%d did not finish", maxKeys)
+			assert.Equal(t, []string{"foo", "foobar", "other", "zed"}, keys, "maxKeys=%d", maxKeys)
+			assert.Equal(t, []string{"foo/", "zed/"}, prefixes, "maxKeys=%d", maxKeys)
+		}
+	})
+
+	// Versioning reaches a prefix object from two directions: a suspended bucket
+	// writes the null version at the key's own path, and a bucket versioned later
+	// finds one already sitting there. Both leave a key that is a directory with
+	// version history beside it.
+	t.Run("Versioned", func(t *testing.T) {
+		setVersioning := func(t *testing.T, bucket, status string) {
+			t.Helper()
+			_, err := cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+				Bucket:                  aws.String(bucket),
+				VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String(status)},
+			})
+			require.NoError(t, err)
+		}
+
+		// Both write orders, in a bucket that is versioned and in one where versioning
+		// was suspended - the suspended one is the case that writes at the key's path.
+		for _, state := range []string{"Enabled", "Suspended"} {
+			bucket := createTestBucket(t, cluster, "test-prefix-"+strings.ToLower(state)+"-")
+			setVersioning(t, bucket, "Enabled")
+			if state == "Suspended" {
+				setVersioning(t, bucket, "Suspended")
+			}
+			put(t, bucket, "child/foo/bar", nested)
+			put(t, bucket, "child/foo", body)
+			put(t, bucket, "prefix/foo", body)
+			put(t, bucket, "prefix/foo/bar", nested)
+
+			assert.Equal(t, []string{"child/foo", "child/foo/bar", "prefix/foo", "prefix/foo/bar"},
+				listKeys(t, bucket), state)
+			for _, key := range []string{"child/foo", "prefix/foo"} {
+				read(t, bucket, key, body)
+			}
+			for _, key := range []string{"child/foo/bar", "prefix/foo/bar"} {
+				read(t, bucket, key, nested)
+			}
+		}
+
+		// A prefix object written before versioning is the key's null version. Removing
+		// that version by id must not take the keys nested under it with it.
+		bucket := createTestBucket(t, cluster, "test-prefix-nullversion-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+		setVersioning(t, bucket, "Enabled")
+		newer := []byte("written after versioning was enabled")
+		versioned, err := cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("collision/foo"),
+			Body:   bytes.NewReader(newer),
+		})
+		require.NoError(t, err)
+
+		for _, v := range []struct {
+			id   string
+			want []byte
+		}{{"null", body}, {aws.StringValue(versioned.VersionId), newer}} {
+			got, err := cluster.s3Client.GetObject(&s3.GetObjectInput{
+				Bucket:    aws.String(bucket),
+				Key:       aws.String("collision/foo"),
+				VersionId: aws.String(v.id),
+			})
+			require.NoError(t, err, "get version %s", v.id)
+			body, err := io.ReadAll(got.Body)
+			require.NoError(t, err)
+			got.Body.Close()
+			assert.Equal(t, v.want, body, "get version %s", v.id)
+		}
+
+		_, err = cluster.s3Client.DeleteObject(&s3.DeleteObjectInput{
+			Bucket:    aws.String(bucket),
+			Key:       aws.String("collision/foo"),
+			VersionId: aws.String("null"),
+		})
+		require.NoError(t, err, "the null version sits on a directory other keys live in")
+
+		read(t, bucket, "collision/foo", newer)
+		read(t, bucket, "collision/foo/bar", nested)
+		remaining, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{
+			Bucket: aws.String(bucket),
+			Prefix: aws.String("collision/foo"),
+		})
+		require.NoError(t, err)
+		for _, v := range remaining.Versions {
+			if aws.StringValue(v.Key) != "collision/foo" {
+				// collision/foo/bar predates versioning too, and keeps its null version.
+				continue
+			}
+			assert.NotEqual(t, "null", aws.StringValue(v.VersionId), "the null version was deleted")
+		}
+	})
+
+	// The two listings walk the tree differently, and a prefix object is the entry
+	// they disagree about: it is a directory the version listing descends through and
+	// a key at the same time. They have to name the same keys and the same prefixes.
+	t.Run("VersionListingMatchesObjectListing", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-parity-")
+		for _, key := range []string{"foo", "foo/bar", "other", "a/foo", "a/foo/bar", "a/z"} {
+			put(t, bucket, key, body)
+		}
+
+		for _, q := range []struct{ prefix, delimiter string }{
+			{"", ""},
+			{"foo/", ""},
+			{"a/", ""},
+			{"a/foo/", ""},
+			{"", "/"},
+			{"a/", "/"},
+		} {
+			name := "prefix=" + q.prefix + " delimiter=" + q.delimiter
+
+			objects, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+				Bucket:    aws.String(bucket),
+				Prefix:    aws.String(q.prefix),
+				Delimiter: aws.String(q.delimiter),
+			})
+			require.NoError(t, err, name)
+
+			versions, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{
+				Bucket:    aws.String(bucket),
+				Prefix:    aws.String(q.prefix),
+				Delimiter: aws.String(q.delimiter),
+			})
+			require.NoError(t, err, name)
+
+			versionKeys := make([]string, 0, len(versions.Versions))
+			for _, v := range versions.Versions {
+				versionKeys = append(versionKeys, aws.StringValue(v.Key))
+			}
+			versionPrefixes := make([]string, 0, len(versions.CommonPrefixes))
+			for _, p := range versions.CommonPrefixes {
+				versionPrefixes = append(versionPrefixes, aws.StringValue(p.Prefix))
+			}
+			sort.Strings(versionKeys)
+			sort.Strings(versionPrefixes)
+
+			objectKeys := collectKeys(objects.Contents)
+			objectPrefixes := collectPrefixes(objects.CommonPrefixes)
+			sort.Strings(objectKeys)
+			sort.Strings(objectPrefixes)
+
+			assert.Equal(t, objectKeys, versionKeys, "keys, %s", name)
+			assert.Equal(t, objectPrefixes, versionPrefixes, "prefixes, %s", name)
+		}
+	})
+
+	// A prefix object written before versioning is the key's null version, so the
+	// version written after it has to take the latest flag off it.
+	t.Run("VersionedAfterPrefixObject", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-versioned-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+		_, err := cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+			Bucket:                  aws.String(bucket),
+			VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String("Enabled")},
+		})
+		require.NoError(t, err)
+		newer := []byte("written after versioning was enabled")
+		put(t, bucket, "collision/foo", newer)
+
+		resp, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{Bucket: aws.String(bucket)})
+		require.NoError(t, err)
+
+		latest := map[string]int{}
+		var nullSize int64 = -1
+		for _, v := range resp.Versions {
+			if aws.BoolValue(v.IsLatest) {
+				latest[aws.StringValue(v.Key)]++
+			}
+			if aws.StringValue(v.Key) == "collision/foo" && aws.StringValue(v.VersionId) == "null" {
+				nullSize = aws.Int64Value(v.Size)
+				assert.False(t, aws.BoolValue(v.IsLatest), "the newer version is the latest one")
+			}
+		}
+		assert.Equal(t, 1, latest["collision/foo"], "exactly one version of a key is the latest")
+		assert.Equal(t, int64(len(body)), nullSize, "the null version keeps the prefix object's size")
+		read(t, bucket, "collision/foo", newer)
+		read(t, bucket, "collision/foo/bar", nested)
+	})
+
+	// A key that other keys are nested under is a copy source and a copy destination
+	// like any other. The keys nested under either end are not part of the copy.
+	t.Run("Copy", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-copy-")
+		copyObject := func(t *testing.T, src, dst string) {
+			t.Helper()
+			_, err := cluster.s3Client.CopyObject(&s3.CopyObjectInput{
+				Bucket:     aws.String(bucket),
+				Key:        aws.String(dst),
+				CopySource: aws.String(bucket + "/" + src),
+			})
+			require.NoError(t, err, "copy %s to %s", src, dst)
+		}
+
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/foo/bar", nested)
+
+		// Out of a prefix object, into a key of its own.
+		copyObject(t, "collision/foo", "plain")
+		read(t, bucket, "plain", body)
+		read(t, bucket, "collision/foo", body)
+		read(t, bucket, "collision/foo/bar", nested)
+
+		// Into a key that other keys are nested under.
+		put(t, bucket, "target/child", nested)
+		copyObject(t, "plain", "target")
+		read(t, bucket, "target", body)
+		read(t, bucket, "target/child", nested)
+
+		// And between two of them.
+		put(t, bucket, "other", []byte("copied between prefix keys"))
+		copyObject(t, "other", "collision/foo")
+		read(t, bucket, "collision/foo", []byte("copied between prefix keys"))
+		read(t, bucket, "collision/foo/bar", nested)
+
+		assert.Equal(t, []string{"collision/foo", "collision/foo/bar", "other", "plain", "target", "target/child"},
+			listKeys(t, bucket))
+	})
+
+	// Rename moves the object off the key without moving the keys nested under it,
+	// which is not what the filer's atomic rename of a directory would do.
+	t.Run("Rename", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-rename-")
+		renameObject := func(t *testing.T, src, dst string) {
+			t.Helper()
+			req, _ := http.NewRequest(http.MethodPut, cluster.s3Endpoint+"/"+bucket+"/"+dst+"?renameObject=", nil)
+			req.Header.Set("x-amz-rename-source", "/"+bucket+"/"+src)
+			signer := v1signer.NewSigner(v1credentials.NewStaticCredentials(testAccessKey, testSecretKey, ""))
+			_, err := signer.Sign(req, nil, "s3", testRegion, time.Now())
+			require.NoError(t, err)
+
+			resp, err := (&http.Client{Timeout: 20 * time.Second}).Do(req)
+			require.NoError(t, err, "rename %s to %s", src, dst)
+			defer resp.Body.Close()
+			io.Copy(io.Discard, resp.Body)
+			require.Equal(t, http.StatusOK, resp.StatusCode, "rename %s to %s", src, dst)
+		}
+
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/foo/bar", nested)
+
+		// Off a prefix object: the key goes, the keys under it stay.
+		renameObject(t, "collision/foo", "moved")
+		read(t, bucket, "moved", body)
+		read(t, bucket, "collision/foo/bar", nested)
+		gone(t, bucket, "collision/foo")
+
+		// Onto a key other keys are nested under.
+		put(t, bucket, "target/child", nested)
+		renameObject(t, "moved", "target")
+		read(t, bucket, "target", body)
+		read(t, bucket, "target/child", nested)
+		gone(t, bucket, "moved")
+
+		assert.Equal(t, []string{"collision/foo/bar", "target", "target/child"}, listKeys(t, bucket))
+	})
+
+	// A directory SeaweedFS keeps its own state in is not a prefix a key can be
+	// stored on: the object would replace that state with its own.
+	t.Run("ReservedDirectory", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-reserved-")
+		_, err := cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+			Bucket:                  aws.String(bucket),
+			VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String("Enabled")},
+		})
+		require.NoError(t, err)
+		put(t, bucket, "foo", body)
+		_, err = cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+			Bucket:                  aws.String(bucket),
+			VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String("Suspended")},
+		})
+		require.NoError(t, err)
+
+		_, err = cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("foo.versions"),
+			Body:   bytes.NewReader(body),
+		})
+		// Not just any error: a permanently impossible write must not come back as a
+		// 500 the SDK retries.
+		var refused awserr.RequestFailure
+		require.ErrorAs(t, err, &refused, "the version history of foo is not a prefix of foo.versions")
+		assert.Equal(t, http.StatusConflict, refused.StatusCode())
+		assert.Equal(t, "ExistingObjectIsDirectory", refused.Code())
+
+		versions, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{Bucket: aws.String(bucket)})
+		require.NoError(t, err)
+		require.Len(t, versions.Versions, 1)
+		assert.Equal(t, "foo", aws.StringValue(versions.Versions[0].Key))
+		read(t, bucket, "foo", body)
+
+		// The multipart staging folder is the other one, and an in-flight upload has
+		// to survive the attempt.
+		staging := createTestBucket(t, cluster, "test-prefix-uploads-")
+		created, err := cluster.s3Client.CreateMultipartUpload(&s3.CreateMultipartUploadInput{
+			Bucket: aws.String(staging),
+			Key:    aws.String("mp.bin"),
+		})
+		require.NoError(t, err)
+
+		_, err = cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(staging),
+			Key:    aws.String(".uploads"),
+			Body:   bytes.NewReader(body),
+		})
+		require.ErrorAs(t, err, &refused, "the multipart staging folder is not a prefix of .uploads")
+		assert.Equal(t, http.StatusConflict, refused.StatusCode())
+		assert.Equal(t, "ExistingObjectIsDirectory", refused.Code())
+
+		uploads, err := cluster.s3Client.ListMultipartUploads(&s3.ListMultipartUploadsInput{Bucket: aws.String(staging)})
+		require.NoError(t, err)
+		require.Len(t, uploads.Uploads, 1)
+		assert.Equal(t, "mp.bin", aws.StringValue(uploads.Uploads[0].Key))
+		_, err = cluster.s3Client.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
+			Bucket:   aws.String(staging),
+			Key:      aws.String("mp.bin"),
+			UploadId: created.UploadId,
+		})
+		require.NoError(t, err)
+	})
+
+	// Either key can be deleted without touching the other.
+	t.Run("Delete", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-delete-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+
+		_, err := cluster.s3Client.DeleteObject(&s3.DeleteObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("collision/foo"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []string{"collision/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo/bar", nested)
+		gone(t, bucket, "collision/foo")
+
+		put(t, bucket, "collision/foo", body)
+		_, err = cluster.s3Client.DeleteObject(&s3.DeleteObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("collision/foo/bar"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []string{"collision/foo"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo", body)
+		gone(t, bucket, "collision/foo/bar")
+	})
+}
diff --git a/weed/filer/filer.go b/weed/filer/filer.go
index 89629e588..a57f5eb3c 100644
--- a/weed/filer/filer.go
+++ b/weed/filer/filer.go
@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
@@ -299,8 +300,12 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry,
 		}
 		glog.V(4).InfofCtx(ctx, "UpdateEntry %s: old entry: %v", entry.FullPath, oldEntry.Name())
 		if err := f.UpdateEntry(ctx, oldEntry, entry); err != nil {
-			glog.ErrorfCtx(ctx, "update entry %s: %v", entry.FullPath, err)
-			return fmt.Errorf("update entry %s: %v", entry.FullPath, err)
+			if errors.Is(err, filer_pb.ErrExistingIsDirectory) || errors.Is(err, filer_pb.ErrExistingIsFile) {
+				glog.V(2).InfofCtx(ctx, "update entry %s: %v", entry.FullPath, err)
+			} else {
+				glog.ErrorfCtx(ctx, "update entry %s: %v", entry.FullPath, err)
+			}
+			return fmt.Errorf("update entry %s: %w", entry.FullPath, err)
 		}
 	}
 
@@ -388,6 +393,16 @@ func (f *Filer) ensureParentDirectoryEntry(ctx context.Context, entry *Entry, di
 		// the original object data remains accessible.
 		glog.V(2).InfofCtx(ctx, "promoting %s from file to directory for %s", dirPath, entry.FullPath)
 		dirEntry.Attr.Mode |= os.ModeDir | 0111
+		// Expiring the entry now deletes the directory row and strands the keys under
+		// it, so the prefix object gives up its lazy TTL. The lifecycle worker still
+		// expires it, through the delete that leaves the directory behind.
+		dirEntry.Attr.TtlSec = 0
+		// An empty object leaves no chunks, content or mime behind, so without the
+		// mark the promotion would hide it.
+		if dirEntry.Extended == nil {
+			dirEntry.Extended = make(map[string][]byte)
+		}
+		dirEntry.Extended[s3_constants.SeaweedFSPrefixObject] = []byte("true")
 		if updateErr := f.Store.UpdateEntry(ctx, dirEntry); updateErr != nil {
 			return fmt.Errorf("promote %s to directory: %v", dirPath, updateErr)
 		}
@@ -481,12 +496,15 @@ func (f *Filer) UpdateEntry(ctx context.Context, oldEntry, entry *Entry) (err er
 		} else {
 			f.ensureEntryInode(entry)
 		}
+		// A type conflict is reported through the sentinel, and callers act on it -
+		// an S3 write of a key other keys are nested under retries as a prefix object -
+		// so it is the caller's outcome that decides whether anything went wrong.
 		if oldEntry.IsDirectory() && !entry.IsDirectory() {
-			glog.ErrorfCtx(ctx, "existing %s is a directory", oldEntry.FullPath)
+			glog.V(2).InfofCtx(ctx, "existing %s is a directory", oldEntry.FullPath)
 			return fmt.Errorf("%s: %w", oldEntry.FullPath, filer_pb.ErrExistingIsDirectory)
 		}
 		if !oldEntry.IsDirectory() && entry.IsDirectory() {
-			glog.ErrorfCtx(ctx, "existing %s is a file", oldEntry.FullPath)
+			glog.V(2).InfofCtx(ctx, "existing %s is a file", oldEntry.FullPath)
 			return fmt.Errorf("%s: %w", oldEntry.FullPath, filer_pb.ErrExistingIsFile)
 		}
 	}
@@ -522,7 +540,9 @@ func (f *Filer) FindEntry(ctx context.Context, p util.FullPath) (entry *Entry, e
 		return Root, nil
 	}
 	entry, err = f.Store.FindEntry(ctx, p)
-	if entry != nil && entry.TtlSec > 0 {
+	// A directory is deleted here one row at a time, which would strand whatever is
+	// under it, so a TTL an older build left on one is not acted on.
+	if entry != nil && entry.TtlSec > 0 && !entry.IsDirectory() {
 		if entry.IsExpireS3Enabled() {
 			if entry.GetS3ExpireTime().Before(time.Now()) && !entry.IsS3Versioning() {
 				if delErr := f.doDeleteEntryMetaAndData(ctx, entry, true, false, nil); delErr != nil {
@@ -561,7 +581,7 @@ func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, sta
 			glog.Errorf("Context is done.")
 			return false, fmt.Errorf("context canceled: %w", ctx.Err())
 		default:
-			if entry.TtlSec > 0 {
+			if entry.TtlSec > 0 && !entry.IsDirectory() {
 				if entry.IsExpireS3Enabled() {
 					if entry.GetS3ExpireTime().Before(time.Now()) && !entry.IsS3Versioning() {
 						// Collect for deletion after iteration completes to avoid DB deadlock
@@ -722,5 +742,6 @@ func (f *Filer) IsDirectoryKeyObject(ctx context.Context, p util.FullPath) (bool
 		return false, nil
 	}
 	// Mirror filer_pb.Entry.IsDirectoryKeyObject so the cleaner keeps a promoted file's data.
-	return entry.IsDirectory() && (entry.Mime != "" || len(entry.GetChunks()) > 0 || len(entry.Content) > 0 || entry.IsInRemoteOnly()), nil
+	_, isPrefixObject := entry.Extended[s3_constants.SeaweedFSPrefixObject]
+	return entry.IsDirectory() && (entry.Mime != "" || len(entry.GetChunks()) > 0 || len(entry.Content) > 0 || entry.IsInRemoteOnly() || isPrefixObject), nil
 }
diff --git a/weed/filer/filer_prefix_object_test.go b/weed/filer/filer_prefix_object_test.go
new file mode 100644
index 000000000..9d1e488c2
--- /dev/null
+++ b/weed/filer/filer_prefix_object_test.go
@@ -0,0 +1,77 @@
+package filer
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestPromoteToPrefixObject covers a key written before the keys nested under it:
+// the file becomes the directory they live in, and has to stay an object of its own.
+func TestPromoteToPrefixObject(t *testing.T) {
+	f, store := newTestFilerWithStubStore()
+	ctx := context.Background()
+
+	object := &Entry{
+		FullPath: util.FullPath("/buckets/bkt/a/foo"),
+		Attr: Attr{
+			Mode:   0o644,
+			Mime:   "text/plain",
+			TtlSec: 3600,
+		},
+		Chunks:   []*filer_pb.FileChunk{{FileId: "1,01", Size: 4}},
+		Extended: map[string][]byte{s3_constants.SeaweedFSExpiresS3: []byte("true")},
+	}
+	require.NoError(t, f.CreateEntry(ctx, object, nil, false, false, nil, false, f.MaxFilenameLength))
+
+	nested := &Entry{FullPath: util.FullPath("/buckets/bkt/a/foo/bar"), Attr: Attr{Mode: 0o644}}
+	require.NoError(t, f.CreateEntry(ctx, nested, nil, false, false, nil, false, f.MaxFilenameLength))
+
+	promoted, err := store.FindEntry(ctx, object.FullPath)
+	require.NoError(t, err)
+	require.NotNil(t, promoted)
+
+	assert.True(t, promoted.IsDirectory(), "the nested key needs a directory here")
+	assert.Equal(t, object.Chunks, promoted.Chunks, "the object's data stays on it")
+	assert.Contains(t, promoted.Extended, s3_constants.SeaweedFSPrefixObject)
+	// Expiring the entry deletes the directory row on its own and strands the keys
+	// under it, so the promotion gives up the lazy TTL.
+	assert.Zero(t, promoted.Attr.TtlSec)
+}
+
+// TestExpiredDirectoryIsNotDeletedOnRead pins the other half: a TTL an older build
+// left on a promoted directory must not take the keys under it with it.
+func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) {
+	f, store := newTestFilerWithStubStore()
+	ctx := context.Background()
+
+	dirPath := util.FullPath("/buckets/bkt/a/foo")
+	expired := time.Now().Add(-2 * time.Hour)
+	require.NoError(t, store.InsertEntry(ctx, &Entry{
+		FullPath: dirPath,
+		Attr: Attr{
+			Mode:   os.ModeDir | 0o755,
+			Crtime: expired,
+			Mtime:  expired,
+			TtlSec: 60,
+		},
+		Extended: map[string][]byte{s3_constants.SeaweedFSExpiresS3: []byte("true")},
+	}))
+	nested := &Entry{FullPath: dirPath + "/bar", Attr: Attr{Mode: 0o644}}
+	require.NoError(t, store.InsertEntry(ctx, nested))
+
+	found, err := f.FindEntry(ctx, dirPath)
+	require.NoError(t, err)
+	require.NotNil(t, found, "deleting it here would leave the nested key unreachable")
+
+	stillThere, err := store.FindEntry(ctx, nested.FullPath)
+	require.NoError(t, err)
+	require.NotNil(t, stillThere)
+}
diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go
index 839c8099a..7174d807e 100644
--- a/weed/pb/filer_pb/filer_client.go
+++ b/weed/pb/filer_pb/filer_client.go
@@ -261,7 +261,7 @@ func DoMkdir(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath
 	glog.V(1).InfofCtx(ctx, "mkdir: %v", request)
 	if err := CreateEntry(ctx, client, request); err != nil {
 		glog.V(0).InfofCtx(ctx, "mkdir %v: %v", request, err)
-		return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
+		return fmt.Errorf("mkdir %s/%s: %w", parentDirectoryPath, dirName, err)
 	}
 
 	return nil
@@ -295,7 +295,7 @@ func MkFile(ctx context.Context, filerClient FilerClient, parentDirectoryPath st
 		glog.V(1).InfofCtx(ctx, "create file: %s/%s", parentDirectoryPath, fileName)
 		if err := CreateEntry(ctx, client, request); err != nil {
 			glog.V(0).InfofCtx(ctx, "create file %v:%v", request, err)
-			return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
+			return fmt.Errorf("create file %s/%s: %w", parentDirectoryPath, fileName, err)
 		}
 
 		return nil
diff --git a/weed/pb/filer_pb/filer_pb_helper.go b/weed/pb/filer_pb/filer_pb_helper.go
index bc8a18ab7..0985c440c 100644
--- a/weed/pb/filer_pb/filer_pb_helper.go
+++ b/weed/pb/filer_pb/filer_pb_helper.go
@@ -26,7 +26,30 @@ func (entry *Entry) IsDirectoryKeyObject() bool {
 	// Also true for a file promoted to a directory by a child write, which keeps its
 	// chunks/content, or its remote entry when the file was tiered to remote storage.
 	return entry.IsDirectory &&
-		((entry.Attributes != nil && entry.Attributes.Mime != "") || len(entry.GetChunks()) > 0 || len(entry.GetContent()) > 0 || entry.IsInRemoteOnly())
+		((entry.Attributes != nil && entry.Attributes.Mime != "") || len(entry.GetChunks()) > 0 || len(entry.GetContent()) > 0 || entry.IsInRemoteOnly() || entry.IsPrefixObject())
+}
+
+// IsPrefixObject reports whether the directory entry also holds the object named by its
+// path without a trailing slash, which an empty object leaves no other trace of.
+func (entry *Entry) IsPrefixObject() bool {
+	if entry == nil || !entry.IsDirectory {
+		return false
+	}
+	_, marked := entry.Extended[s3_constants.SeaweedFSPrefixObject]
+	return marked
+}
+
+// MarkPrefixObject turns a file entry into the directory entry that stores it, so keys
+// nested under its key can be created beside it.
+func (entry *Entry) MarkPrefixObject() {
+	entry.IsDirectory = true
+	if entry.Attributes != nil {
+		entry.Attributes.FileMode |= uint32(os.ModeDir) | 0111
+	}
+	if entry.Extended == nil {
+		entry.Extended = make(map[string][]byte)
+	}
+	entry.Extended[s3_constants.SeaweedFSPrefixObject] = []byte("true")
 }
 
 func (entry *Entry) GetExpiryTime() (expiryTime int64) {
diff --git a/weed/pb/filer_pb/filer_pb_helper_test.go b/weed/pb/filer_pb/filer_pb_helper_test.go
index 92b3eac48..b8e696d74 100644
--- a/weed/pb/filer_pb/filer_pb_helper_test.go
+++ b/weed/pb/filer_pb/filer_pb_helper_test.go
@@ -1,7 +1,10 @@
 package filer_pb
 
 import (
+	"os"
 	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
 
 func TestIsDirectoryKeyObject(t *testing.T) {
@@ -20,6 +23,7 @@ func TestIsDirectoryKeyObject(t *testing.T) {
 		{"directory with chunks and nil attributes", &Entry{IsDirectory: true, Chunks: chunk}, true},
 		{"regular file with chunks", &Entry{IsDirectory: false, Attributes: &FuseAttributes{}, Chunks: chunk}, false},
 		{"remote mount directory has no remote size", &Entry{IsDirectory: true, Attributes: &FuseAttributes{}, RemoteEntry: &RemoteEntry{StorageName: "s3"}}, false},
+		{"empty prefix object has only the mark", &Entry{IsDirectory: true, Attributes: &FuseAttributes{}, Extended: map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}}, true},
 	}
 
 	for _, c := range cases {
@@ -30,3 +34,28 @@ func TestIsDirectoryKeyObject(t *testing.T) {
 		})
 	}
 }
+
+func TestMarkPrefixObject(t *testing.T) {
+	entry := &Entry{Name: "foo", Attributes: &FuseAttributes{FileMode: 0644}}
+
+	if entry.IsPrefixObject() {
+		t.Fatal("a file is not a prefix object")
+	}
+
+	entry.MarkPrefixObject()
+
+	if !entry.IsDirectory || !entry.IsPrefixObject() {
+		t.Errorf("MarkPrefixObject() left %+v", entry)
+	}
+	// The filer derives the entry type from the mode, so the directory bit has to be
+	// on it for the store to keep this as a directory.
+	if mode := os.FileMode(entry.Attributes.FileMode); mode&os.ModeDir == 0 || mode.Perm()&0111 != 0111 {
+		t.Errorf("FileMode = %v, want a traversable directory", mode)
+	}
+
+	// A directory the mark was stripped from is a plain directory again.
+	delete(entry.Extended, s3_constants.SeaweedFSPrefixObject)
+	if entry.IsPrefixObject() || entry.IsDirectoryKeyObject() {
+		t.Error("a demoted directory names no key")
+	}
+}
diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go
index be4633e9e..74a9680a0 100644
--- a/weed/s3api/filer_util.go
+++ b/weed/s3api/filer_util.go
@@ -24,7 +24,18 @@ func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn fun
 
 func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk, fn func(entry *filer_pb.Entry)) error {
 
-	return filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, fn)
+	err := filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, fn)
+	if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(fileName) {
+		// Other keys are nested under this one, so the object goes onto the directory
+		// they live in - the same place a PutObject of this key writes it.
+		err = filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, func(entry *filer_pb.Entry) {
+			if fn != nil {
+				fn(entry)
+			}
+			entry.MarkPrefixObject()
+		})
+	}
+	return err
 
 }
 
@@ -187,6 +198,10 @@ func clearDirectoryMarkerMetadata(entry *filer_pb.Entry) {
 	filtered := make(map[string][]byte)
 	for k, v := range entry.Extended {
 		lowerKey := strings.ToLower(k)
+		if lowerKey == s3_constants.SeaweedFSPrefixObject {
+			// The path is a plain directory again, not a key of its own.
+			continue
+		}
 		if strings.HasPrefix(lowerKey, "xattr-") || strings.HasPrefix(lowerKey, s3_constants.SeaweedFSInternalPrefix) {
 			filtered[k] = v
 		}
diff --git a/weed/s3api/filer_util_delete_test.go b/weed/s3api/filer_util_delete_test.go
index 31339a470..68389daa0 100644
--- a/weed/s3api/filer_util_delete_test.go
+++ b/weed/s3api/filer_util_delete_test.go
@@ -96,6 +96,36 @@ func TestDeleteObjectEntryDemotesNonEmptyDirectoryMarker(t *testing.T) {
 	}, updated.Extended)
 }
 
+// A prefix object is demoted the same way, and the mark has to go with the data:
+// the path is a plain directory again, and no longer a key of its own.
+func TestDeleteObjectEntryDemotesPrefixObject(t *testing.T) {
+	client := &deleteObjectEntryTestClient{
+		deleteResp: &filer_pb.DeleteEntryResponse{
+			Error: filer.MsgFailDelNonEmptyFolder + ": /buckets/test/photos",
+		},
+		lookupResp: &filer_pb.LookupDirectoryEntryResponse{
+			Entry: &filer_pb.Entry{
+				Name:        "photos",
+				IsDirectory: true,
+				Attributes:  &filer_pb.FuseAttributes{},
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSPrefixObject: []byte("true"),
+					s3_constants.ExtETagKey:            []byte("etag"),
+				},
+			},
+		},
+	}
+
+	require.NoError(t, deleteObjectEntry(client, "/buckets/test", "photos", true, false))
+	require.NotNil(t, client.updateReq)
+
+	updated := client.updateReq.Entry
+	require.NotNil(t, updated)
+	assert.False(t, updated.IsPrefixObject())
+	assert.False(t, updated.IsDirectoryKeyObject())
+	assert.Empty(t, updated.Extended)
+}
+
 func TestDeleteObjectEntryTreatsImplicitDirectoryAsSuccessfulNoop(t *testing.T) {
 	client := &deleteObjectEntryTestClient{
 		deleteResp: &filer_pb.DeleteEntryResponse{
diff --git a/weed/s3api/s3_constants/header.go b/weed/s3api/s3_constants/header.go
index aa13c6df5..e7fe61b77 100644
--- a/weed/s3api/s3_constants/header.go
+++ b/weed/s3api/s3_constants/header.go
@@ -139,6 +139,12 @@ const (
 	// SeaweedFS internal metadata prefix (used to filter internal headers from client responses)
 	SeaweedFSInternalPrefix = "x-seaweedfs-"
 
+	// SeaweedFSPrefixObject marks a directory entry that also holds the object named by
+	// its path without a trailing slash. S3 keys are flat, so "a/b" and "a/b/c" are
+	// independent keys, and the filer stores the first one on the directory the second
+	// one lives under.
+	SeaweedFSPrefixObject = "x-seaweedfs-prefix-object"
+
 	// SeaweedFS internal metadata keys for encryption (prefixed to avoid automatic HTTP header conversion)
 	SeaweedFSSSEKMSKey = "x-seaweedfs-sse-kms-key" // Key for storing serialized SSE-KMS metadata
 	SeaweedFSSSES3Key  = "x-seaweedfs-sse-s3-key"  // Key for storing serialized SSE-S3 metadata
diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go
index d9340ac29..43c5d75db 100644
--- a/weed/s3api/s3api_object_handlers.go
+++ b/weed/s3api/s3api_object_handlers.go
@@ -364,7 +364,7 @@ func (s3a *S3ApiServer) hasChildren(ctx context.Context, bucket, prefix string)
 // Such a path is not an S3 object: GET/HEAD answer 404 like AWS does for a prefix, and
 // Hadoop-style clients then discover the directory through their LIST fallback.
 func isBareDirectory(entry *filer_pb.Entry) bool {
-	return entry != nil && entry.IsDirectory && filer.FileSize(entry) == 0
+	return entry != nil && entry.IsDirectory && filer.FileSize(entry) == 0 && !entry.IsPrefixObject()
 }
 
 // checkDirectoryObject checks if the object is a directory object (ends with "/") and if it exists
diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go
index a704e8957..f165c04ac 100644
--- a/weed/s3api/s3api_object_handlers_copy.go
+++ b/weed/s3api/s3api_object_handlers_copy.go
@@ -10,6 +10,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"reflect"
 	"strconv"
 	"strings"
@@ -69,6 +70,23 @@ func hasPrefixFold(s, prefix string) bool {
 	return len(s) >= len(prefix) && strings.EqualFold(s[:len(prefix)], prefix)
 }
 
+// prefixObjectSource presents the object stored on a key that other keys are nested
+// under as the plain object that key names. The entry itself stays a directory - it
+// is where those keys live - so a caller that copies or moves the object works off
+// this view and leaves the directory, and everything under it, alone.
+func prefixObjectSource(entry *filer_pb.Entry) *filer_pb.Entry {
+	if entry == nil || !entry.IsPrefixObject() {
+		return entry
+	}
+	flattened := proto.Clone(entry).(*filer_pb.Entry)
+	flattened.IsDirectory = false
+	if flattened.Attributes != nil {
+		flattened.Attributes.FileMode &^= uint32(os.ModeDir)
+	}
+	delete(flattened.Extended, s3_constants.SeaweedFSPrefixObject)
+	return flattened
+}
+
 // classifyCopySourceError maps a copy-source lookup to an S3 error: a missing
 // or directory source is NoSuchKey like AWS, but a transient store error
 // becomes a retryable 5xx so a resumable copy/commit survives a blip.
@@ -180,6 +198,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 	}
 
 	entry, err := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState)
+	entry = prefixObjectSource(entry)
 	if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
 		return
@@ -250,6 +269,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 		routeInPlace := owner != "" && !mimeChanged
 		selfCopyBody := func() s3err.ErrorCode {
 			currentEntry, currentErr := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState)
+			currentEntry = prefixObjectSource(currentEntry)
 			if errCode := classifyCopySourceError(currentEntry, currentErr); errCode != s3err.ErrNone {
 				return errCode
 			}
diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go
index bd60d05ef..78e7794c7 100644
--- a/weed/s3api/s3api_object_handlers_delete.go
+++ b/weed/s3api/s3api_object_handlers_delete.go
@@ -299,11 +299,10 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 					}
 				}
 				if versionId == "null" {
-					deleteCode = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass)
+					deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass)
 				} else {
-					deleteCode = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass)
+					deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass), true
 				}
-				deleteHandled = true
 			}
 		}
 	}
diff --git a/weed/s3api/s3api_object_handlers_dir_test.go b/weed/s3api/s3api_object_handlers_dir_test.go
index ff7057094..163d084da 100644
--- a/weed/s3api/s3api_object_handlers_dir_test.go
+++ b/weed/s3api/s3api_object_handlers_dir_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
 
 func TestServeDirectoryContentContentType(t *testing.T) {
@@ -47,3 +48,31 @@ func TestServeDirectoryContentContentType(t *testing.T) {
 		})
 	}
 }
+
+func TestIsBareDirectory(t *testing.T) {
+	tests := []struct {
+		name  string
+		entry *filer_pb.Entry
+		want  bool
+	}{
+		{"nothing there", nil, false},
+		{"file", &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{}}, false},
+		{"directory a listing stands for", &filer_pb.Entry{IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}, true},
+		{"promoted file with data", &filer_pb.Entry{IsDirectory: true, Content: []byte("data"), Attributes: &filer_pb.FuseAttributes{FileSize: 4}}, false},
+		{
+			// The key is real even at zero bytes, so it answers rather than 404s.
+			name: "empty prefix object",
+			entry: &filer_pb.Entry{IsDirectory: true, Attributes: &filer_pb.FuseAttributes{},
+				Extended: map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isBareDirectory(tt.entry); got != tt.want {
+				t.Errorf("isBareDirectory() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/weed/s3api/s3api_object_handlers_list.go b/weed/s3api/s3api_object_handlers_list.go
index 0b8de645f..f06fedebd 100644
--- a/weed/s3api/s3api_object_handlers_list.go
+++ b/weed/s3api/s3api_object_handlers_list.go
@@ -400,6 +400,28 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 			return ""
 		}
 
+		// addCommonPrefix records a CommonPrefix the page has not emitted yet. maxKeys is
+		// unsigned, so the decrement is clamped rather than allowed to wrap: a prefix
+		// object spends two slots on one entry, and the budget is only checked between
+		// entries.
+		addCommonPrefix := func(prefix string) {
+			for i := range commonPrefixes {
+				if commonPrefixes[i].Prefix == prefix {
+					if prefix == lastCommonPrefix {
+						lastPrefixConfirmed = true
+					}
+					return
+				}
+			}
+			commonPrefixes = append(commonPrefixes, PrefixEntry{Prefix: prefix})
+			if cursor.maxKeys > 0 {
+				cursor.maxKeys--
+			}
+			lastEntryWasCommonPrefix = true
+			lastCommonPrefix = prefix
+			lastPrefixNullBackers, lastPrefixConfirmed = nil, true
+		}
+
 		retractPrefixBacking := func(prefix, dir, name string) {
 			if prefix != lastCommonPrefix || !lastPrefixNullBackers[dir+"/"+name] {
 				return
@@ -514,6 +536,29 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 							}
 						}
 					}
+					// A prefix object's key carries no trailing slash, so it lists as a
+					// plain key. The traversal never descends into it under a delimiter,
+					// so the CommonPrefix its nested keys fold into is added here too.
+					if entry.IsPrefixObject() {
+						key := fmt.Sprintf("%s/%s", dir, entry.Name)[len(bucketPrefix):]
+						if strings.HasPrefix(key, originalPrefix) {
+							if folded := prefixForKey(dir, entry.Name); folded != "" {
+								// The key and everything under it fold into the same prefix.
+								addCommonPrefix(folded)
+								return
+							}
+							appendOrDedup(newListEntry(s3a, entry, "", dirName, entryName, bucketPrefix, fetchOwner, false, false))
+							lastEntryWasCommonPrefix = false
+						}
+						// The key and the prefix come off one entry, which a marker names as a
+						// whole, so a page ending between them cannot resume at the prefix:
+						// a marker of "<key>/" means the subtree is done, not that it is
+						// next. The page runs one item over maxKeys instead of dropping it.
+						if childPrefix := prefixForKey(dir, entry.Name+"/"); childPrefix != "" && s3a.hasChildren(ctx, bucket, key) {
+							addCommonPrefix(childPrefix)
+						}
+						return
+					}
 					// When delimiter is specified, apply delimiter logic to directory key objects too
 					if delimiter != "" && entry.IsDirectoryKeyObject() {
 						// Apply the same delimiter logic as for regular files
diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go
index e83fd35b1..67aead988 100644
--- a/weed/s3api/s3api_object_handlers_put.go
+++ b/weed/s3api/s3api_object_handlers_put.go
@@ -876,7 +876,14 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 				Entry:     entry,
 			}
 			glog.V(3).Infof("putToFiler: Calling CreateEntry for %s", filePath)
-			if err := filer_pb.CreateEntry(context.Background(), client, req); err != nil {
+			err := filer_pb.CreateEntry(context.Background(), client, req)
+			if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) {
+				// Other keys are nested under this one. S3 keys are flat, so the key is
+				// stored on the directory they live under rather than refused.
+				entry.MarkPrefixObject()
+				err = filer_pb.CreateEntry(context.Background(), client, req)
+			}
+			if err != nil {
 				glog.Errorf("putToFiler: CreateEntry returned error: %v", err)
 				return err
 			}
@@ -1192,6 +1199,14 @@ func (s3a *S3ApiServer) setSSEResponseHeaders(w http.ResponseWriter, r *http.Req
 	}
 }
 
+// isReservedDirectoryName reports whether a directory standing at an object's path is
+// one SeaweedFS keeps its own state in - a multipart staging folder, or a key's version
+// history - rather than a prefix the key can be stored on. Writing the object onto it
+// would replace that state with the object's own.
+func isReservedDirectoryName(name string) bool {
+	return name == s3_constants.MultipartUploadsFolder || strings.HasSuffix(name, s3_constants.VersionsFolder)
+}
+
 func filerErrorToS3Error(err error) s3err.ErrorCode {
 	if err == nil {
 		return s3err.ErrNone
diff --git a/weed/s3api/s3api_object_handlers_rename.go b/weed/s3api/s3api_object_handlers_rename.go
index 8ac6c0be6..a2275c89a 100644
--- a/weed/s3api/s3api_object_handlers_rename.go
+++ b/weed/s3api/s3api_object_handlers_rename.go
@@ -94,6 +94,8 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque
 
 	errCode = s3a.withRenameWriteLocks(bucket, srcObject, dstObject, func() s3err.ErrorCode {
 		entry, err := s3a.resolveCopySourceEntry(bucket, srcObject, "", "")
+		srcIsPrefixObject := entry.IsPrefixObject()
+		entry = prefixObjectSource(entry)
 		if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone {
 			return errCode
 		}
@@ -103,7 +105,7 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque
 		if errCode := s3a.checkConditionalHeaders(r, bucket, dstObject); errCode != s3err.ErrNone {
 			return errCode
 		}
-		return s3a.renameObjectEntry(r.Context(), bucket, srcObject, dstObject)
+		return s3a.renameObjectEntry(r.Context(), bucket, srcObject, dstObject, entry, srcIsPrefixObject)
 	})
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
@@ -218,19 +220,28 @@ func (s3a *S3ApiServer) withRenameWriteLocks(bucket, srcObject, dstObject string
 	})
 }
 
-func (s3a *S3ApiServer) renameObjectEntry(ctx context.Context, bucket, srcObject, dstObject string) s3err.ErrorCode {
+func (s3a *S3ApiServer) renameObjectEntry(ctx context.Context, bucket, srcObject, dstObject string, srcEntry *filer_pb.Entry, srcIsPrefixObject bool) s3err.ErrorCode {
 	srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, srcObject)).DirAndName()
 	dstDir, dstName := util.FullPath(s3a.toFilerPath(bucket, dstObject)).DirAndName()
 
-	// The move overwrites an existing destination object, but a directory in
-	// the way is a conflict the filer reports as an opaque error.
-	if existing, err := s3a.getEntry(dstDir, dstName); err == nil && existing.IsDirectory {
-		return s3err.ErrExistingObjectIsDirectory
-	} else if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
+	// The move overwrites an existing destination object. A directory there is not a
+	// conflict: it means other keys are nested under the destination key, and the
+	// object goes onto the directory they live in, the way a PutObject of that key
+	// would put it there.
+	dstHoldsNestedKeys := false
+	if existing, err := s3a.getEntry(dstDir, dstName); err == nil {
+		dstHoldsNestedKeys = existing.IsDirectory
+	} else if !errors.Is(err, filer_pb.ErrNotFound) {
 		glog.Errorf("RenameObject %s: destination %s: %v", bucket, dstObject, err)
 		return s3err.ErrInternalError
 	}
 
+	// AtomicRenameEntry moves a directory by moving everything under it, and the keys
+	// nested under either end of this rename are not part of what is being renamed.
+	if srcIsPrefixObject || dstHoldsNestedKeys {
+		return s3a.renameKeyHoldingNestedKeys(bucket, srcObject, dstObject, srcEntry)
+	}
+
 	err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		_, err := client.AtomicRenameEntry(ctx, &filer_pb.AtomicRenameEntryRequest{
 			OldDirectory: srcDir,
@@ -249,3 +260,38 @@ func (s3a *S3ApiServer) renameObjectEntry(ctx context.Context, bucket, srcObject
 	}
 	return s3err.ErrNone
 }
+
+// renameKeyHoldingNestedKeys moves an object when either key of the rename is one
+// other keys are nested under. Such a key is stored on the directory those keys live
+// in, which has to stay where it is, so the object's own data is written at the
+// destination and then stripped off the source key - the entry survives as the plain
+// directory it also is. Both keys are held under their write locks for the whole
+// move, so no other S3 write interleaves; a crash between the two steps leaves the
+// destination written and the source still there, which a retry settles.
+func (s3a *S3ApiServer) renameKeyHoldingNestedKeys(bucket, srcObject, dstObject string, srcEntry *filer_pb.Entry) s3err.ErrorCode {
+	dstPath := util.FullPath(s3a.toFilerPath(bucket, dstObject))
+	dstDir, dstName := dstPath.DirAndName()
+
+	chunks, err := s3a.copyChunks(srcEntry, string(dstPath))
+	if err != nil {
+		glog.Errorf("RenameObject %s: copy chunks of %s: %v", bucket, srcObject, err)
+		return s3err.ErrInternalError
+	}
+
+	if err := s3a.mkFile(dstDir, dstName, chunks, func(entry *filer_pb.Entry) {
+		copyEntryToTarget(entry, srcEntry)
+		entry.Chunks = chunks
+	}); err != nil {
+		glog.Errorf("RenameObject %s: write %s: %v", bucket, dstObject, err)
+		s3a.deleteOrphanedChunks(chunks)
+		return filerErrorToS3Error(err)
+	}
+
+	// The destination holds copies now, so the source's own chunks go with it.
+	srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, srcObject)).DirAndName()
+	if err := s3a.rmObject(srcDir, srcName, true, false); err != nil {
+		glog.Errorf("RenameObject %s: strip %s: %v", bucket, srcObject, err)
+		return s3err.ErrInternalError
+	}
+	return s3err.ErrNone
+}
diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go
index ccd18d68f..c17ea43a4 100644
--- a/weed/s3api/s3api_object_versioned_finalize.go
+++ b/weed/s3api/s3api_object_versioned_finalize.go
@@ -156,8 +156,10 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck
 // routedDeleteNullVersion deletes the null version (the regular object entry, not
 // a .versions file) off the distributed lock. There is no pointer to recompute;
 // the WORM guards, when present, gate the delete on the object entry itself
-// (condition defaults to lock_key).
-func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass bool) s3err.ErrorCode {
+// (condition defaults to lock_key). The second return reports whether the delete
+// was settled here: the raw delete cannot remove an entry other keys are nested
+// under, which the lock path handles by stripping the object off it instead.
+func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass bool) (s3err.ErrorCode, bool) {
 	fullpath := util.NewFullPath(s3a.bucketDir(bucket), object)
 	dir, name := fullpath.DirAndName()
 	resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
@@ -170,15 +172,15 @@ func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket,
 	})
 	switch {
 	case err != nil:
-		glog.Errorf("routedDeleteNullVersion: %s/%s on %s: %v", bucket, object, owner, err)
-		return s3err.ErrInternalError
+		glog.Warningf("routedDeleteNullVersion: %s/%s on %s, falling back to lock: %v", bucket, object, owner, err)
+		return s3err.ErrNone, false
 	case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED:
-		return s3err.ErrAccessDenied
+		return s3err.ErrAccessDenied, true
 	case resp.Error != "":
-		glog.Errorf("routedDeleteNullVersion: %s/%s: %s", bucket, object, resp.Error)
-		return s3err.ErrInternalError
+		glog.Warningf("routedDeleteNullVersion: %s/%s returned %q, falling back to lock", bucket, object, resp.Error)
+		return s3err.ErrNone, false
 	default:
-		return s3err.ErrNone
+		return s3err.ErrNone, true
 	}
 }
 
diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go
index 62f447bd9..5e1028b1d 100644
--- a/weed/s3api/s3api_object_versioning.go
+++ b/weed/s3api/s3api_object_versioning.go
@@ -22,6 +22,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -856,6 +857,13 @@ func (vc *versionCollector) collectVersions(currentPath, relativePath string) er
 							vc.commonPrefixes[commonPrefix] = true
 						}
 
+						// The prefix rolled up here belongs to the keys nested under this
+						// entry. A prefix object is also a key of its own, and that key
+						// carries no trailing slash, so it does not roll up with them.
+						if entry.IsPrefixObject() {
+							vc.processPrefixObject(currentPath, entryPath, entry)
+						}
+
 						// Skip further processing (recursion or addition) for this entry
 						// because it has been rolled up into the CommonPrefix
 						continue
@@ -879,13 +887,31 @@ func (vc *versionCollector) collectVersions(currentPath, relativePath string) er
 	return nil
 }
 
+// processPrefixObject emits the null version of a key that other keys are nested
+// under. The key carries no trailing slash, so it is a null object like any other and
+// needs the same reconciliation against a .versions sibling - but the directory it is
+// stored on is walked as an ancestor of the requested prefix, which its own key does
+// not have to match.
+func (vc *versionCollector) processPrefixObject(currentPath, entryPath string, entry *filer_pb.Entry) {
+	if !strings.HasPrefix(entryPath, vc.prefix) {
+		return
+	}
+	if vc.delimiter != "" && strings.Contains(entryPath[len(vc.prefix):], vc.delimiter) {
+		// The key folds into the same CommonPrefix its nested keys do.
+		return
+	}
+	vc.processRegularFile(currentPath, entryPath, entry)
+}
+
 // processDirectory handles directory entries
 func (vc *versionCollector) processDirectory(currentPath, entryPath string, entry *filer_pb.Entry) error {
 	// Handle explicit S3 directory object. Match ListObjectsV2's
 	// IsDirectoryKeyObject (any non-empty mime), not just FolderMimeType:
 	// an SDK PutObject of "dir/" carries a default Content-Type, so the two
 	// listings must agree on what counts as a directory key.
-	if entry.IsDirectoryKeyObject() {
+	if entry.IsPrefixObject() {
+		vc.processPrefixObject(currentPath, entryPath, entry)
+	} else if entry.IsDirectoryKeyObject() {
 		vc.processExplicitDirectory(entryPath, entry)
 	}
 
@@ -1113,8 +1139,11 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket,
 			return nil
 		}
 
-		// Delete the regular file
-		deleteErr := s3a.rmObject(bucketDir, normalizedObject, !metadataOnly, false)
+		// Delete the regular file. rmObject takes a parent and a name, and the demote
+		// it falls back to for an entry other keys are nested under writes the entry
+		// back under that parent - so a key with a slash in it has to be split first.
+		dir, name := util.NewFullPath(bucketDir, normalizedObject).DirAndName()
+		deleteErr := s3a.rmObject(dir, name, !metadataOnly, false)
 		if deleteErr != nil {
 			// Check if file was already deleted by another process
 			if _, checkErr := s3a.getEntry(bucketDir, normalizedObject); checkErr != nil {