diff --git a/.github/workflows/s3-go-tests.yml b/.github/workflows/s3-go-tests.yml
index b5aeb8634..fcdc5bb3d 100644
--- a/.github/workflows/s3-go-tests.yml
+++ b/.github/workflows/s3-go-tests.yml
@@ -245,6 +245,51 @@ jobs:
           path: test/s3/retention/weed-test*.log
           retention-days: 3
 
+  s3-lifecycle-tests:
+    name: S3 Lifecycle Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 15
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+        id: go
+
+      - name: Install SeaweedFS
+        run: |
+          go install -buildvcs=false
+
+      - name: Run S3 Lifecycle Tests
+        timeout-minutes: 12
+        working-directory: test/s3/lifecycle
+        run: |
+          set -x
+          make test-with-server
+
+      - name: Show server logs on failure
+        if: failure()
+        working-directory: test/s3/lifecycle
+        run: |
+          if [ -f weed-test.log ]; then
+            echo "=== Last 200 lines of server logs ==="
+            tail -200 weed-test.log
+          fi
+          ps aux | grep -E "(weed|test)" || true
+          netstat -tlnp 2>/dev/null | grep -E "(8333|9333|8080|8888)" || true
+
+      - name: Upload test logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: s3-lifecycle-test-logs
+          path: test/s3/lifecycle/weed-test*.log
+          retention-days: 3
+
   s3-checksum-tests:
     name: S3 Checksum Tests
     runs-on: ubuntu-22.04
diff --git a/test/s3/lifecycle/Makefile b/test/s3/lifecycle/Makefile
new file mode 100644
index 000000000..26612ca76
--- /dev/null
+++ b/test/s3/lifecycle/Makefile
@@ -0,0 +1,80 @@
+# S3 Lifecycle Test Makefile
+# End-to-end test of the event-driven lifecycle worker, driven by
+# the s3.lifecycle.run-shard shell command.
+
+.PHONY: help build-weed start-server stop-server test test-with-server clean health-check
+
+WEED_BINARY := ../../../weed/weed_binary
+S3_PORT := 8333
+MASTER_PORT := 9333
+VOLUME_PORT := 8080
+FILER_PORT := 8888
+ACCESS_KEY ?= some_access_key1
+SECRET_KEY ?= some_secret_key1
+TEST_TIMEOUT := 10m
+TEST_PATTERN := TestLifecycle
+SERVER_DIR := ./test-volume-data/server-data
+
+help:
+	@echo "S3 Lifecycle Test Makefile"
+	@echo ""
+	@echo "Targets:"
+	@echo "  build-weed       - build the SeaweedFS binary"
+	@echo "  start-server     - start a local 'weed mini' cluster"
+	@echo "  stop-server      - stop the cluster"
+	@echo "  test             - run tests against an already-running cluster"
+	@echo "  test-with-server - start, run tests, stop"
+	@echo "  health-check     - is the S3 endpoint up?"
+	@echo "  clean            - remove test artifacts"
+
+build-weed:
+	@echo "Building SeaweedFS binary..."
+	@cd ../../../weed && go build -o weed_binary .
+	@chmod +x $(WEED_BINARY)
+
+start-server: build-weed
+	@echo "Starting weed mini..."
+	@rm -f weed-server.pid
+	@mkdir -p $(SERVER_DIR)
+	@AWS_ACCESS_KEY_ID=$(ACCESS_KEY) AWS_SECRET_ACCESS_KEY=$(SECRET_KEY) $(WEED_BINARY) mini \
+		-dir=$(SERVER_DIR) \
+		-s3.port=$(S3_PORT) \
+		> weed-test.log 2>&1 & \
+		echo $$! > weed-server.pid
+	@for i in $$(seq 1 90); do \
+		if curl -s http://localhost:$(S3_PORT) >/dev/null 2>&1; then \
+			echo "server up after $$i s"; exit 0; \
+		fi; \
+		sleep 1; \
+	done; \
+	echo "server did not start within 90s"; \
+	tail -50 weed-test.log; \
+	exit 1
+
+stop-server:
+	@if [ -f weed-server.pid ]; then \
+		PID=$$(cat weed-server.pid); \
+		kill -TERM $$PID 2>/dev/null || true; \
+		sleep 1; \
+		kill -KILL $$PID 2>/dev/null || true; \
+		rm -f weed-server.pid; \
+	fi
+
+health-check:
+	@curl -s http://localhost:$(S3_PORT) >/dev/null && echo "S3 up on $(S3_PORT)" || (echo "S3 not reachable"; exit 1)
+
+test:
+	@WEED_BINARY=$$(cd ../../../weed && pwd)/weed_binary \
+	S3_ENDPOINT=http://localhost:$(S3_PORT) \
+	S3_GRPC_ENDPOINT=localhost:$$(($(S3_PORT) + 10000)) \
+	MASTER_ENDPOINT=http://localhost:$(MASTER_PORT) \
+	FILER_GRPC_ADDRESS=localhost:$$(($(FILER_PORT) + 10000)) \
+	go test -v -timeout $(TEST_TIMEOUT) -run $(TEST_PATTERN)
+
+test-with-server: start-server
+	@$(MAKE) test || (RC=$$?; $(MAKE) stop-server; exit $$RC)
+	@$(MAKE) stop-server
+
+clean: stop-server
+	@rm -rf $(SERVER_DIR) test-volume-data
+	@rm -f weed-test.log weed-server.pid
diff --git a/test/s3/lifecycle/README.md b/test/s3/lifecycle/README.md
new file mode 100644
index 000000000..aafe327d5
--- /dev/null
+++ b/test/s3/lifecycle/README.md
@@ -0,0 +1,38 @@
+# S3 Lifecycle Integration Tests
+
+End-to-end test of the event-driven S3 lifecycle worker, exercised through
+the `s3.lifecycle.run-shard` shell command.
+
+## Why backdate mtimes?
+
+The S3 API rejects `Expiration.Days < 1`, so a literal "wait one day"
+integration test isn't workable. Each test sets up a 1-day expiration rule,
+puts the target object, then rewrites its filer entry's `Mtime` to ~30 days
+ago via `filer_pb.UpdateEntry`. From the engine's perspective the object is
+past its expiration window the moment the shell command starts.
+
+## Running
+
+```sh
+# build the binary, start a local mini cluster, run tests, stop it
+make test-with-server
+
+# or, if a cluster is already running on the default ports
+make test
+```
+
+The test runs the shell command once with `-shards 0-15` (one filer
+subscription covering all 16 shards) rather than computing the target
+object's shard up front. This keeps the test independent of the
+`ShardID(bucket, key)` hash function — only that some shard reaches the
+deletion within the polling window.
+
+## Environment
+
+| variable             | default                  | description                |
+|----------------------|--------------------------|----------------------------|
+| `WEED_BINARY`        | _required_               | path to `weed_binary`      |
+| `S3_ENDPOINT`        | `http://localhost:8333`  | S3 API URL                 |
+| `S3_GRPC_ENDPOINT`   | `localhost:18333`        | S3 gRPC for lifecycle dispatch |
+| `MASTER_ENDPOINT`    | `http://localhost:9333`  | master HTTP                |
+| `FILER_GRPC_ADDRESS` | `localhost:18888`        | filer gRPC for `UpdateEntry` |
diff --git a/test/s3/lifecycle/s3_lifecycle_test.go b/test/s3/lifecycle/s3_lifecycle_test.go
new file mode 100644
index 000000000..29cac7445
--- /dev/null
+++ b/test/s3/lifecycle/s3_lifecycle_test.go
@@ -0,0 +1,233 @@
+// Package lifecycle is the end-to-end test for the event-driven S3
+// lifecycle worker, driven by the s3.lifecycle.run-shard shell command.
+//
+// The S3 API rejects Expiration.Days < 1, so a literal "wait one day"
+// integration test is unworkable. Instead, the test backdates the target
+// object's mtime via the filer's UpdateEntry RPC: from the engine's
+// perspective the object is past its expiration window the moment the
+// shell command starts, and the dispatcher fires immediately.
+//
+// Variables (set by the Makefile):
+//
+//	WEED_BINARY         - path to the built `weed` binary
+//	S3_ENDPOINT         - http://host:port for the S3 API
+//	FILER_GRPC_ADDRESS  - host:port of the filer's gRPC listener
+package lifecycle
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/url"
+	"os"
+	"os/exec"
+	"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/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+const (
+	defaultS3Endpoint     = "http://localhost:8333"
+	defaultS3GrpcEndpoint = "localhost:18333"
+	defaultMasterEndpt    = "http://localhost:9333"
+	defaultFilerGRPC      = "localhost:18888"
+	bucketLifecycleXMLKey = "s3-bucket-lifecycle-configuration-xml"
+	bucketsPath           = "/buckets"
+	accessKey             = "some_access_key1"
+	secretKey           = "some_secret_key1"
+	region              = "us-east-1"
+)
+
+func envOr(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+func s3Client(t *testing.T) *s3.Client {
+	t.Helper()
+	endpoint := envOr("S3_ENDPOINT", defaultS3Endpoint)
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion(region),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")),
+		config.WithEndpointResolverWithOptions(aws.EndpointResolverWithOptionsFunc(
+			func(service, region string, _ ...interface{}) (aws.Endpoint, error) {
+				return aws.Endpoint{URL: endpoint, SigningRegion: region, HostnameImmutable: true}, nil
+			})),
+	)
+	require.NoError(t, err)
+	return s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = true })
+}
+
+func filerClient(t *testing.T) (filer_pb.SeaweedFilerClient, func()) {
+	t.Helper()
+	addr := envOr("FILER_GRPC_ADDRESS", defaultFilerGRPC)
+	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	return filer_pb.NewSeaweedFilerClient(conn), func() { conn.Close() }
+}
+
+// uniqueBucket returns a fresh bucket name; the shell command picks up all
+// buckets with lifecycle config, so two parallel-running tests would conflict
+// on cursor state. Each test gets its own bucket.
+func uniqueBucket(prefix string) string {
+	return fmt.Sprintf("lc-%s-%d", prefix, time.Now().UnixNano())
+}
+
+func mustCreateBucket(t *testing.T, c *s3.Client, name string) {
+	t.Helper()
+	_, err := c.CreateBucket(context.Background(), &s3.CreateBucketInput{Bucket: aws.String(name)})
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		// Best effort: empty + delete.
+		listOut, _ := c.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{Bucket: aws.String(name)})
+		if listOut != nil {
+			for _, o := range listOut.Contents {
+				c.DeleteObject(context.Background(), &s3.DeleteObjectInput{Bucket: aws.String(name), Key: o.Key})
+			}
+		}
+		c.DeleteBucket(context.Background(), &s3.DeleteBucketInput{Bucket: aws.String(name)})
+	})
+}
+
+func putExpirationLifecycle(t *testing.T, c *s3.Client, bucket, prefix string, days int32) {
+	t.Helper()
+	_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
+		Bucket: aws.String(bucket),
+		LifecycleConfiguration: &types.BucketLifecycleConfiguration{
+			Rules: []types.LifecycleRule{
+				{
+					ID:     aws.String("expire-prefix"),
+					Status: types.ExpirationStatusEnabled,
+					Filter: &types.LifecycleRuleFilter{Prefix: aws.String(prefix)},
+					Expiration: &types.LifecycleExpiration{
+						Days: aws.Int32(days),
+					},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+}
+
+func putObject(t *testing.T, c *s3.Client, bucket, key, body string) {
+	t.Helper()
+	_, err := c.PutObject(context.Background(), &s3.PutObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+		Body:   strings.NewReader(body),
+	})
+	require.NoError(t, err)
+}
+
+// backdateMtime rewrites the object's filer entry attributes so its Mtime
+// is daysOld days in the past. This sidesteps the AWS-spec minimum of one
+// day for Expiration.Days, letting the lifecycle dispatcher fire on demand.
+func backdateMtime(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket, key string, daysOld int) {
+	t.Helper()
+	dir, name := splitBucketKey(bucket, key)
+	resp, err := fc.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+		Directory: dir,
+		Name:      name,
+	})
+	require.NoError(t, err, "lookup %s/%s", dir, name)
+	require.NotNil(t, resp.Entry)
+	require.NotNil(t, resp.Entry.Attributes)
+
+	resp.Entry.Attributes.Mtime = time.Now().Add(-time.Duration(daysOld) * 24 * time.Hour).Unix()
+	resp.Entry.Attributes.MtimeNs = 0
+	_, err = fc.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
+		Directory: dir,
+		Entry:     resp.Entry,
+	})
+	require.NoError(t, err, "update %s/%s", dir, name)
+}
+
+func splitBucketKey(bucket, key string) (dir, name string) {
+	full := bucketsPath + "/" + bucket + "/" + key
+	if i := strings.LastIndex(full, "/"); i >= 0 {
+		return full[:i], full[i+1:]
+	}
+	return full, ""
+}
+
+// runShellCommand invokes `weed shell` with a one-shot s3.lifecycle.run-shard
+// command piped via stdin and returns the combined stdout+stderr.
+func runShellCommand(t *testing.T, command string) string {
+	t.Helper()
+	binary := envOr("WEED_BINARY", "")
+	require.NotEmpty(t, binary, "WEED_BINARY must be set")
+	masterEndpoint := envOr("MASTER_ENDPOINT", defaultMasterEndpt)
+	master := strings.TrimPrefix(masterEndpoint, "http://")
+	if u, err := url.Parse(masterEndpoint); err == nil && u.Host != "" {
+		master = u.Host
+	}
+	cmd := exec.Command(binary, "shell", "-master="+master)
+	cmd.Stdin = strings.NewReader(command + "\nexit\n")
+	var out bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &out, &out
+	err := cmd.Run()
+	output := out.String()
+	if err != nil {
+		t.Logf("shell command output:\n%s", output)
+		t.Fatalf("shell exec failed: %v", err)
+	}
+	return output
+}
+
+// TestLifecycleExpirationFiresOnBackdatedObject is the end-to-end
+// validation: a 1-day expiration rule plus an object whose mtime has been
+// backdated to 30 days ago must result in deletion when the shell command
+// runs the matching shard.
+func TestLifecycleExpirationFiresOnBackdatedObject(t *testing.T) {
+	c := s3Client(t)
+	fc, fcClose := filerClient(t)
+	defer fcClose()
+
+	bucket := uniqueBucket("expire")
+	mustCreateBucket(t, c, bucket)
+	putExpirationLifecycle(t, c, bucket, "expire/", 1)
+
+	const oldKey = "expire/old.txt"
+	const freshKey = "keep/fresh.txt"
+	putObject(t, c, bucket, oldKey, "old")
+	putObject(t, c, bucket, freshKey, "fresh")
+	backdateMtime(t, fc, bucket, oldKey, 30)
+
+	// One subscription handles every shard via -shards 0-15; this stays
+	// independent of which (bucket, key) hash lands the target on.
+	// -runtime caps the run by wall-clock so the subprocess exits even
+	// when the in-shard event count never reaches -events.
+	out := runShellCommand(t, fmt.Sprintf(
+		"s3.lifecycle.run-shard -shards 0-15 -s3 %s -events 0 -dispatch 200ms -checkpoint 5s -runtime 10s",
+		envOr("S3_GRPC_ENDPOINT", defaultS3GrpcEndpoint),
+	))
+	t.Logf("shell command output:\n%s", out)
+	require.NotContains(t, out, "FATAL", "shell output:\n%s", out)
+
+	// Allow the dispatched delete to land in the filer + propagate to S3.
+	require.Eventuallyf(t, func() bool {
+		_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket), Key: aws.String(oldKey),
+		})
+		return err != nil
+	}, 30*time.Second, 500*time.Millisecond, "expected %s/%s to be deleted", bucket, oldKey)
+
+	// The fresh, in-prefix-but-recent object must remain (1d rule, mtime
+	// is now). Best-evidence check after waiting for the same poll window.
+	_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
+		Bucket: aws.String(bucket), Key: aws.String(freshKey),
+	})
+	require.NoError(t, err, "%s should still exist", freshKey)
+}
diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go
index 13e48a9cc..e4e38094f 100644
--- a/weed/s3api/s3api_internal_lifecycle.go
+++ b/weed/s3api/s3api_internal_lifecycle.go
@@ -3,15 +3,13 @@ package s3api
 import (
 	"bytes"
 	"context"
-	"crypto/sha256"
 	"errors"
-	"sort"
-	"strconv"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 )
 
 // LifecycleDelete executes one (rule, action) verdict: re-fetch, identity
@@ -36,8 +34,7 @@ func (s3a *S3ApiServer) LifecycleDelete(ctx context.Context, req *s3_lifecycle_p
 		return retryLater("TRANSPORT_ERROR: " + err.Error()), nil
 	}
 
-	live := computeEntryIdentity(entry)
-	if !identityMatches(live, req.ExpectedIdentity) {
+	if !identityMatches(computeEntryIdentity(entry), req.ExpectedIdentity) {
 		return noopResolved("STALE_IDENTITY"), nil
 	}
 
@@ -142,36 +139,12 @@ func computeEntryIdentity(entry *filer_pb.Entry) *s3_lifecycle_pb.EntryIdentity
 		id.Size = int64(entry.Attributes.FileSize)
 	}
 	if len(entry.GetChunks()) > 0 {
-		id.HeadFid = entry.GetChunks()[0].FileId
+		id.HeadFid = entry.GetChunks()[0].GetFileIdString()
 	}
-	id.ExtendedHash = hashExtended(entry.Extended)
+	id.ExtendedHash = s3lifecycle.HashExtended(entry.Extended)
 	return id
 }
 
-// hashExtended is length-prefixed; a forged tag value can't collide with a
-// legitimate multi-tag map.
-func hashExtended(ext map[string][]byte) []byte {
-	if len(ext) == 0 {
-		return nil
-	}
-	keys := make([]string, 0, len(ext))
-	for k := range ext {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-	h := sha256.New()
-	for _, k := range keys {
-		h.Write([]byte(strconv.Itoa(len(k))))
-		h.Write([]byte{':'})
-		h.Write([]byte(k))
-		v := ext[k]
-		h.Write([]byte(strconv.Itoa(len(v))))
-		h.Write([]byte{':'})
-		h.Write(v)
-	}
-	return h.Sum(nil)
-}
-
 func identityMatches(live, want *s3_lifecycle_pb.EntryIdentity) bool {
 	if want == nil {
 		// No CAS witness (early bootstrap); skip.
diff --git a/weed/s3api/s3api_internal_lifecycle_test.go b/weed/s3api/s3api_internal_lifecycle_test.go
index c7975f716..7d7ded78a 100644
--- a/weed/s3api/s3api_internal_lifecycle_test.go
+++ b/weed/s3api/s3api_internal_lifecycle_test.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 )
 
 func TestComputeEntryIdentity_BasicFields(t *testing.T) {
@@ -45,7 +46,7 @@ func TestComputeEntryIdentity_NilSafeMissingChunks(t *testing.T) {
 func TestHashExtended_OrderStable(t *testing.T) {
 	a := map[string][]byte{"k1": []byte("v1"), "k2": []byte("v2")}
 	b := map[string][]byte{"k2": []byte("v2"), "k1": []byte("v1")}
-	if !bytes.Equal(hashExtended(a), hashExtended(b)) {
+	if !bytes.Equal(s3lifecycle.HashExtended(a), s3lifecycle.HashExtended(b)) {
 		t.Fatalf("hash should be insensitive to map iteration order")
 	}
 }
@@ -55,16 +56,16 @@ func TestHashExtended_DelimiterCollisionResistant(t *testing.T) {
 	// Length-prefix encoding must keep them apart.
 	a := map[string][]byte{"k1": []byte("v1"), "k2": []byte("v2")}
 	b := map[string][]byte{"k1": []byte("v1k2v2")}
-	if bytes.Equal(hashExtended(a), hashExtended(b)) {
+	if bytes.Equal(s3lifecycle.HashExtended(a), s3lifecycle.HashExtended(b)) {
 		t.Fatalf("delimiter-forged Extended payloads must not collide")
 	}
 }
 
 func TestHashExtended_NilEqualsEmpty(t *testing.T) {
-	if got := hashExtended(nil); len(got) != 0 {
+	if got := s3lifecycle.HashExtended(nil); len(got) != 0 {
 		t.Fatalf("nil should produce zero-length hash, got %d bytes", len(got))
 	}
-	if got := hashExtended(map[string][]byte{}); len(got) != 0 {
+	if got := s3lifecycle.HashExtended(map[string][]byte{}); len(got) != 0 {
 		t.Fatalf("empty map should produce zero-length hash, got %d bytes", len(got))
 	}
 }
diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go
index 6de626b9b..a232347f3 100644
--- a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go
+++ b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go
@@ -200,8 +200,9 @@ func toProtoIdentity(id *router.EntryIdentity) *s3_lifecycle_pb.EntryIdentity {
 		return nil
 	}
 	return &s3_lifecycle_pb.EntryIdentity{
-		MtimeNs: id.MtimeNs,
-		Size:    id.Size,
-		HeadFid: id.HeadFid,
+		MtimeNs:      id.MtimeNs,
+		Size:         id.Size,
+		HeadFid:      id.HeadFid,
+		ExtendedHash: id.ExtendedHash,
 	}
 }
diff --git a/weed/s3api/s3lifecycle/dispatcher/pipeline.go b/weed/s3api/s3lifecycle/dispatcher/pipeline.go
index c376af4a5..75027550a 100644
--- a/weed/s3api/s3lifecycle/dispatcher/pipeline.go
+++ b/weed/s3api/s3lifecycle/dispatcher/pipeline.go
@@ -9,18 +9,31 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
-// Pipeline composes the per-shard reader, router, dispatcher, and cursor
-// checkpoint into a single Run loop. One Pipeline per (worker, shard).
+// Pipeline composes the reader, router, dispatcher, and cursor checkpoint
+// into a single Run loop. One Pipeline can handle a contiguous shard span
+// or any explicit set of shards via Shards; ShardID still works for the
+// single-shard case (and is preferred for short-form configuration).
+//
+// Internally there is exactly one filer subscription regardless of how
+// many shards Shards contains; events are filtered by the reader's
+// ShardPredicate and routed to the matching shard's Cursor + Schedule
+// inside the existing dispatch goroutine — no per-shard goroutines.
 type Pipeline struct {
-	ShardID     int
+	ShardID     int   // used when Shards is empty
+	Shards      []int // overrides ShardID when non-empty
 	BucketsPath string
 
-	Engine    *engine.Engine
+	Engine *engine.Engine
+	// Cursor is consulted only when len(Shards) <= 1. Range mode allocates
+	// a fresh Cursor per shard internally.
 	Cursor    *reader.Cursor
 	Persister reader.Persister
 	Client    LifecycleClient
@@ -47,46 +60,88 @@ const (
 	defaultDispatchTick   = 5 * time.Second
 	defaultCheckpointTick = 30 * time.Second
 	defaultEventBuffer    = 1024
+	shutdownDrainTimeout  = 30 * time.Second
+	shutdownSaveTimeout   = 5 * time.Second
 )
 
-// Run blocks until ctx is canceled or a fatal error occurs. On exit, the
-// cursor is persisted; in-flight schedule entries are dropped (the meta-log
-// is the durable buffer, so a restart re-derives them).
+// shardState bundles per-shard mutable state so the single dispatch
+// goroutine can route an event to the right cursor + schedule by lookup.
+type shardState struct {
+	cursor   *reader.Cursor
+	dispatch *Dispatcher
+}
+
+// Run blocks until ctx is canceled or a fatal error occurs. On exit, every
+// shard's cursor is persisted; in-flight schedule entries are dropped
+// (the meta-log is the durable buffer, so a restart re-derives them).
 func (p *Pipeline) Run(ctx context.Context) error {
-	if p.Engine == nil || p.Cursor == nil || p.Persister == nil ||
-		p.Client == nil || p.FilerClient == nil {
+	if p.Engine == nil || p.Persister == nil || p.Client == nil || p.FilerClient == nil {
 		return errors.New("pipeline: missing required dependency")
 	}
 	if p.BucketsPath == "" {
 		return errors.New("pipeline: BucketsPath required")
 	}
 
-	// Restore cursor; freezes re-arm naturally when the reader re-encounters
-	// the poison event at MinTsNs and the dispatch state machine drives it
+	// Resolve the active shard set. Single-shard configurations populate
+	// either Shards=[N] or Shards=nil with ShardID=N (latter is the legacy
+	// path that also supplies a Cursor); both feed the same range model.
+	shardIDs := p.Shards
+	if len(shardIDs) == 0 {
+		shardIDs = []int{p.ShardID}
+	}
+	shardSet := make(map[int]struct{}, len(shardIDs))
+	for _, s := range shardIDs {
+		if s < 0 || s >= s3lifecycle.ShardCount {
+			return fmt.Errorf("pipeline: shard %d out of [0,%d)", s, s3lifecycle.ShardCount)
+		}
+		shardSet[s] = struct{}{}
+	}
+
+	// Per-shard cursor + dispatcher. Cursors restore from the durable
+	// store; freezes re-arm naturally when the reader re-encounters the
+	// poison event at MinTsNs and the dispatch state machine drives it
 	// back to BLOCKED.
-	state, err := p.Persister.Load(ctx, p.ShardID)
-	if err != nil {
-		return fmt.Errorf("cursor load: %w", err)
+	states := make(map[int]*shardState, len(shardIDs))
+	var minStartTsNs int64 = -1
+	for _, shardID := range shardIDs {
+		c := p.Cursor
+		if len(shardIDs) != 1 || c == nil {
+			c = reader.NewCursor()
+		}
+		state, err := p.Persister.Load(ctx, shardID)
+		if err != nil {
+			return fmt.Errorf("cursor load shard=%d: %w", shardID, err)
+		}
+		c.Restore(state)
+		states[shardID] = &shardState{
+			cursor: c,
+			dispatch: &Dispatcher{
+				ShardID:  shardID,
+				Client:   p.Client,
+				Cursor:   c,
+				Schedule: router.NewSchedule(),
+			},
+		}
+		if mt := c.MinTsNs(); mt > 0 && (minStartTsNs < 0 || mt < minStartTsNs) {
+			minStartTsNs = mt
+		}
 	}
-	p.Cursor.Restore(state)
-
-	dispatch := &Dispatcher{
-		ShardID:  p.ShardID,
-		Client:   p.Client,
-		Cursor:   p.Cursor,
-		Schedule: router.NewSchedule(),
+	if minStartTsNs < 0 {
+		minStartTsNs = 0
 	}
 
-	// 2. Wire reader -> router -> schedule via a buffered channel.
 	bufSize := p.EventBuffer
 	if bufSize <= 0 {
 		bufSize = defaultEventBuffer
 	}
 	events := make(chan *reader.Event, bufSize)
 	rd := &reader.Reader{
-		ShardID:     p.ShardID,
 		BucketsPath: p.BucketsPath,
-		Cursor:      p.Cursor,
+		ShardPredicate: func(s int) bool {
+			_, ok := shardSet[s]
+			return ok
+		},
+		StartTsNs:   minStartTsNs,
 		Events:      events,
 		EventBudget: p.EventBudget,
 	}
@@ -104,13 +159,16 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		defer wg.Done()
 		defer close(events)
 		readerErr = rd.Run(runCtx, p.FilerClient, p.ClientName, p.ClientID)
-		if readerErr != nil && !errors.Is(readerErr, context.Canceled) {
-			glog.Errorf("lifecycle reader: shard=%d: %v", p.ShardID, readerErr)
+		if readerErr != nil && !isCtxShutdown(readerErr) {
+			glog.Errorf("lifecycle reader: shards=%v: %v", shardIDs, readerErr)
 		}
 		cancel() // wake the dispatcher goroutine to drain & exit
 	}()
 
-	// Router/dispatcher goroutine: pulls events, routes them, ticks schedule.
+	// Router/dispatcher goroutine: pulls events, routes them to per-shard
+	// schedules, ticks every shard's dispatcher on the same cadence, and
+	// checkpoints every shard's cursor on the checkpoint cadence. One
+	// goroutine handles all shards — there is no fan-out per shard.
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -128,28 +186,46 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		defer ct.Stop()
 		snap := p.Engine.Snapshot()
 
+		drainAll := func() {
+			drainCtx, drainCancel := context.WithTimeout(context.Background(), shutdownDrainTimeout)
+			defer drainCancel()
+			now := time.Now()
+			for _, st := range states {
+				st.dispatch.Tick(drainCtx, now)
+			}
+		}
+
 		for {
 			select {
 			case <-runCtx.Done():
-				dispatch.Tick(context.Background(), time.Now())
+				drainAll()
 				return
 			case ev, ok := <-events:
 				if !ok {
-					dispatch.Tick(context.Background(), time.Now())
+					drainAll()
 					return
 				}
+				st := states[ev.ShardID]
+				if st == nil {
+					continue
+				}
 				if snap == nil {
 					snap = p.Engine.Snapshot()
 				}
 				for _, m := range router.Route(snap, ev, time.Now()) {
-					dispatch.Schedule.Add(m)
+					st.dispatch.Schedule.Add(m)
 				}
 			case <-dt.C:
-				snap = p.Engine.Snapshot() // refresh between tick boundaries
-				dispatch.Tick(runCtx, time.Now())
+				snap = p.Engine.Snapshot()
+				now := time.Now()
+				for _, st := range states {
+					st.dispatch.Tick(runCtx, now)
+				}
 			case <-ct.C:
-				if err := p.Persister.Save(runCtx, p.ShardID, p.Cursor.Snapshot()); err != nil {
-					glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", p.ShardID, err)
+				for shardID, st := range states {
+					if err := p.Persister.Save(runCtx, shardID, st.cursor.Snapshot()); err != nil {
+						glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", shardID, err)
+					}
 				}
 			}
 		}
@@ -158,12 +234,33 @@ func (p *Pipeline) Run(ctx context.Context) error {
 	wg.Wait()
 
 	// Final cursor checkpoint on graceful shutdown.
-	if err := p.Persister.Save(context.Background(), p.ShardID, p.Cursor.Snapshot()); err != nil {
-		glog.Warningf("lifecycle cursor final save: shard=%d: %v", p.ShardID, err)
+	for shardID, st := range states {
+		saveCtx, saveCancel := context.WithTimeout(context.Background(), shutdownSaveTimeout)
+		err := p.Persister.Save(saveCtx, shardID, st.cursor.Snapshot())
+		saveCancel()
+		if err != nil {
+			glog.Warningf("lifecycle cursor final save: shard=%d: %v", shardID, err)
+		}
 	}
 
-	if readerErr != nil && !errors.Is(readerErr, context.Canceled) {
+	if readerErr != nil && !isCtxShutdown(readerErr) {
 		return readerErr
 	}
 	return nil
 }
+
+// isCtxShutdown reports whether err is a graceful ctx-driven shutdown
+// (Canceled or DeadlineExceeded), including the gRPC status forms that
+// don't unwrap to the std-lib ctx errors.
+func isCtxShutdown(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	if code := status.Code(err); code == codes.Canceled || code == codes.DeadlineExceeded {
+		return true
+	}
+	return false
+}
diff --git a/weed/s3api/s3lifecycle/identity.go b/weed/s3api/s3lifecycle/identity.go
new file mode 100644
index 000000000..8d7d166d6
--- /dev/null
+++ b/weed/s3api/s3lifecycle/identity.go
@@ -0,0 +1,36 @@
+package s3lifecycle
+
+import (
+	"crypto/sha256"
+	"sort"
+	"strconv"
+)
+
+// HashExtended is the canonical fingerprint of an entry's Extended map for
+// use in lifecycle identity-CAS. Length-prefixed so a forged tag value
+// can't collide with a legitimate multi-tag map. Returns nil for empty.
+//
+// Both sides of the LifecycleDelete CAS — the worker that captures the
+// schedule-time identity and the server that re-fetches the live entry —
+// must call this same function so the bytes match exactly.
+func HashExtended(ext map[string][]byte) []byte {
+	if len(ext) == 0 {
+		return nil
+	}
+	keys := make([]string, 0, len(ext))
+	for k := range ext {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	h := sha256.New()
+	for _, k := range keys {
+		h.Write([]byte(strconv.Itoa(len(k))))
+		h.Write([]byte{':'})
+		h.Write([]byte(k))
+		v := ext[k]
+		h.Write([]byte(strconv.Itoa(len(v))))
+		h.Write([]byte{':'})
+		h.Write(v)
+	}
+	return h.Sum(nil)
+}
diff --git a/weed/s3api/s3lifecycle/reader/reader.go b/weed/s3api/s3lifecycle/reader/reader.go
index 1e0ab1f3e..a5eb1c382 100644
--- a/weed/s3api/s3lifecycle/reader/reader.go
+++ b/weed/s3api/s3lifecycle/reader/reader.go
@@ -17,6 +17,7 @@ type Event struct {
 	TsNs      int64
 	Bucket    string
 	Key       string
+	ShardID   int
 	OldEntry  *filer_pb.Entry
 	NewEntry  *filer_pb.Entry
 	NewParent string
@@ -32,15 +33,24 @@ func (e *Event) IsCreate() bool {
 	return e.OldEntry == nil && e.NewEntry != nil
 }
 
-// Reader subscribes to the filer meta-log for one shard. It owns a Cursor
-// (per-ActionKey position within the shard) and emits in-shard Events to a
-// channel; the downstream router consumes events and ack-advances the cursor
-// for matched ActionKeys when their actions complete.
+// Reader subscribes to the filer meta-log and emits in-range Events to a
+// channel. One subscription handles a contiguous span (or arbitrary set)
+// of shards via ShardPredicate; the downstream router/dispatcher consume
+// events and ack-advance the per-shard cursor for matched ActionKeys
+// when their actions complete.
 type Reader struct {
-	ShardID     int    // [0, s3lifecycle.ShardCount)
+	// ShardID and ShardPredicate are alternatives — set at most one.
+	// ShardPredicate wins if both are populated.
+	ShardID        int             // [0, s3lifecycle.ShardCount); used when ShardPredicate is nil
+	ShardPredicate func(int) bool  // accepts an event when true; nil falls back to ShardID equality
+
 	BucketsPath string // e.g. "/buckets"
-	Cursor      *Cursor
-	Events      chan<- *Event
+	// Cursor is the single-shard cursor used for SinceNs when StartTsNs is 0.
+	// Range callers pass StartTsNs directly and leave Cursor nil; SinceNs=0
+	// then means "subscribe from the start of the meta-log".
+	Cursor    *Cursor
+	StartTsNs int64
+	Events    chan<- *Event
 
 	// EventBudget caps how many events Run processes before returning nil.
 	// Zero = unbounded; the run continues until ctx cancellation or stream
@@ -53,15 +63,15 @@ type Reader struct {
 	bucketsPathSlash string
 }
 
-// Run subscribes via SubscribeMetadata starting at Cursor.MinTsNs(), filters
-// to this shard, and emits Events. Returns on ctx.Done(), io.EOF, or
-// stream error. Caller is responsible for closing Events if it owns it.
+// Run subscribes via SubscribeMetadata starting at the configured position,
+// filters to the configured shard set, and emits Events. Returns on
+// ctx.Done(), io.EOF, or stream error. Caller is responsible for closing
+// Events if it owns it.
 func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, clientName string, clientID int32) error {
-	if r.ShardID < 0 || r.ShardID >= s3lifecycle.ShardCount {
-		return fmt.Errorf("reader: shard_id %d out of range", r.ShardID)
-	}
-	if r.Cursor == nil {
-		return errors.New("reader: nil Cursor")
+	if r.ShardPredicate == nil {
+		if r.ShardID < 0 || r.ShardID >= s3lifecycle.ShardCount {
+			return fmt.Errorf("reader: shard_id %d out of range and no ShardPredicate", r.ShardID)
+		}
 	}
 	if r.Events == nil {
 		return errors.New("reader: nil Events channel")
@@ -74,10 +84,14 @@ func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, cl
 		r.bucketsPathSlash += "/"
 	}
 
+	sinceNs := r.StartTsNs
+	if sinceNs == 0 && r.Cursor != nil {
+		sinceNs = r.Cursor.MinTsNs()
+	}
 	stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
 		ClientName:             clientName,
 		PathPrefix:             r.BucketsPath,
-		SinceNs:                r.Cursor.MinTsNs(),
+		SinceNs:                sinceNs,
 		ClientId:               clientID,
 		ClientSupportsBatching: true,
 	})
@@ -118,7 +132,12 @@ func (r *Reader) dispatchOne(ctx context.Context, resp *filer_pb.SubscribeMetada
 	if !ok {
 		return nil
 	}
-	if s3lifecycle.ShardID(bucket, key) != r.ShardID {
+	shardID := s3lifecycle.ShardID(bucket, key)
+	if r.ShardPredicate != nil {
+		if !r.ShardPredicate(shardID) {
+			return nil
+		}
+	} else if shardID != r.ShardID {
 		return nil
 	}
 
@@ -126,6 +145,7 @@ func (r *Reader) dispatchOne(ctx context.Context, resp *filer_pb.SubscribeMetada
 		TsNs:      resp.TsNs,
 		Bucket:    bucket,
 		Key:       key,
+		ShardID:   shardID,
 		OldEntry:  resp.EventNotification.OldEntry,
 		NewEntry:  resp.EventNotification.NewEntry,
 		NewParent: resp.EventNotification.NewParentPath,
@@ -212,6 +232,14 @@ func (r *Reader) extractBucketKey(resp *filer_pb.SubscribeMetadataResponse) (str
 // LogStartup is a small helper for callers that want a one-line readable
 // description of where the reader is starting.
 func (r *Reader) LogStartup() {
+	sinceNs := r.StartTsNs
+	if sinceNs == 0 && r.Cursor != nil {
+		sinceNs = r.Cursor.MinTsNs()
+	}
+	if r.ShardPredicate != nil {
+		glog.V(1).Infof("lifecycle reader: shard=range sinceNs=%d budget=%d", sinceNs, r.EventBudget)
+		return
+	}
 	glog.V(1).Infof("lifecycle reader: shard=%d sinceNs=%d budget=%d",
-		r.ShardID, r.Cursor.MinTsNs(), r.EventBudget)
+		r.ShardID, sinceNs, r.EventBudget)
 }
diff --git a/weed/s3api/s3lifecycle/router/router.go b/weed/s3api/s3lifecycle/router/router.go
index 9f50fe402..9badf77e4 100644
--- a/weed/s3api/s3lifecycle/router/router.go
+++ b/weed/s3api/s3lifecycle/router/router.go
@@ -31,9 +31,10 @@ type Match struct {
 // s3_lifecycle_pb.EntryIdentity but stay in-package so the router doesn't
 // pull a proto dependency.
 type EntryIdentity struct {
-	MtimeNs int64
-	Size    int64
-	HeadFid string
+	MtimeNs      int64
+	Size         int64
+	HeadFid      string
+	ExtendedHash []byte
 }
 
 // Route returns the matches that fire for ev against snap. Only EVENT_DRIVEN
@@ -70,11 +71,13 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match {
 		if action.Mode != engine.ModeEventDriven {
 			continue
 		}
-		// Evaluate at the scheduled dispatch time, not the event time:
-		// ExpirationDays gates on now >= modtime + N, which is exactly
-		// what holds when we actually dispatch. The dispatcher's
-		// identity-CAS catches drift if the object changes meanwhile.
-		dueTime := eventTime.Add(action.Delay)
+		// Schedule from ModTime, not the meta-log event time: a backdated
+		// or out-of-band entry update has eventTime ≈ now but ModTime far
+		// in the past, so eventTime+Delay would push the dispatch into the
+		// future even though the rule fires immediately. ModTime+Delay is
+		// the correct fire moment; the dispatcher's identity-CAS catches
+		// drift if the object changes meanwhile.
+		dueTime := info.ModTime.Add(action.Delay)
 		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
 		if res.Action == s3lifecycle.ActionNone {
 			continue
@@ -138,8 +141,12 @@ func buildIdentity(ev *reader.Event) *EntryIdentity {
 		id.Size = int64(entry.Attributes.FileSize)
 	}
 	if len(entry.GetChunks()) > 0 {
-		id.HeadFid = entry.GetChunks()[0].FileId
+		// Meta-log events arrive with chunk.FileId cleared by
+		// BeforeEntrySerialization; GetFileIdString reconstructs it from
+		// Fid so the worker matches the server-side fingerprint.
+		id.HeadFid = entry.GetChunks()[0].GetFileIdString()
 	}
+	id.ExtendedHash = s3lifecycle.HashExtended(entry.Extended)
 	return id
 }
 
diff --git a/weed/s3api/s3lifecycle/router/router_test.go b/weed/s3api/s3lifecycle/router/router_test.go
index 51038ce28..7d5e57551 100644
--- a/weed/s3api/s3lifecycle/router/router_test.go
+++ b/weed/s3api/s3lifecycle/router/router_test.go
@@ -195,3 +195,34 @@ func TestRouteIdentityCapturedForNewEntry(t *testing.T) {
 		t.Fatalf("MtimeNs=%d, want %d (Mtime*1e9)", id.MtimeNs, wantNs)
 	}
 }
+
+func TestRouteIdentityHashesExtended(t *testing.T) {
+	// A normal S3 PUT stores ETag/content-type etc. in Extended; the worker
+	// must hash these into ExtendedHash so the server's identity-CAS sees
+	// the same fingerprint. Without this the live entry's non-nil
+	// ExtendedHash diverges from the worker's nil value and every
+	// dispatch returns NOOP_RESOLVED:STALE_IDENTITY.
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWith(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.Add(-48 * time.Hour)
+	ev := eventCreate("bk", "k", old.Unix(), 1, old.UnixNano())
+	ev.NewEntry.Extended = map[string][]byte{
+		"X-Amz-Meta-Etag": []byte("abc123"),
+		"Content-Type":    []byte("text/plain"),
+	}
+
+	matches := Route(snap, ev, now)
+	if len(matches) != 1 {
+		t.Fatalf("expected 1 match, got %v", matches)
+	}
+	id := matches[0].Identity
+	if id == nil || len(id.ExtendedHash) == 0 {
+		t.Fatalf("ExtendedHash not captured for non-empty Extended: %+v", id)
+	}
+	want := s3lifecycle.HashExtended(ev.NewEntry.Extended)
+	if string(id.ExtendedHash) != string(want) {
+		t.Fatalf("ExtendedHash mismatch:\n got %x\nwant %x", id.ExtendedHash, want)
+	}
+}
diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go
new file mode 100644
index 000000000..334491eb0
--- /dev/null
+++ b/weed/shell/command_s3_lifecycle_run_shard.go
@@ -0,0 +1,376 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/lifecycle_xml"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3LifecycleRunShard{})
+}
+
+type commandS3LifecycleRunShard struct{}
+
+func (c *commandS3LifecycleRunShard) Name() string {
+	return "s3.lifecycle.run-shard"
+}
+
+func (c *commandS3LifecycleRunShard) Help() string {
+	return `manually run one or more shards of the event-driven S3 lifecycle worker
+
+Subscribes once to the filer meta-log, filters events to the configured
+(bucket, key-prefix-hash) shards, routes them through the compiled lifecycle
+engine, and dispatches due actions to the S3 server's LifecycleDelete RPC.
+Persists each shard's cursor to /etc/s3/lifecycle/cursors/shard-NN.json so
+subsequent runs resume.
+
+The -shards form covers a range or set; one filer subscription handles the
+whole set, with no per-shard goroutine fan-out. Provide either -shard or
+-shards, not both.
+
+	# single shard
+	s3.lifecycle.run-shard -shard 0 -s3 localhost:8333 -events 100
+
+	# contiguous range, all 16 shards via one subscription
+	s3.lifecycle.run-shard -shards 0-15 -s3 localhost:8333 -events 5000
+
+	# explicit set
+	s3.lifecycle.run-shard -shards 0,3,7 -s3 localhost:8333
+
+	# custom cadence
+	s3.lifecycle.run-shard -shards 0-15 -s3 s3-host:8333 -dispatch 1s -checkpoint 10s
+`
+}
+
+func (c *commandS3LifecycleRunShard) HasTag(CommandTag) bool { return false }
+
+func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	shard := fs.Int("shard", -1, "single shard id in [0, 16); use -shards for a range or set")
+	shardsSpec := fs.String("shards", "", "shard range \"lo-hi\" or comma list \"a,b,c\"; mutually exclusive with -shard")
+	s3Endpoint := fs.String("s3", "", "s3 server gRPC endpoint, host:port")
+	eventBudget := fs.Int("events", 1000, "max in-shard events to process before returning (0 = unbounded; counts only events that pass the shard filter)")
+	dispatchTick := fs.Duration("dispatch", 5*time.Second, "dispatcher tick cadence")
+	checkpointTick := fs.Duration("checkpoint", 30*time.Second, "cursor checkpoint cadence")
+	runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout. -events alone can hang on quiet shards")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	shards, err := resolveShardSelection(*shard, *shardsSpec)
+	if err != nil {
+		return err
+	}
+	if *s3Endpoint == "" {
+		return fmt.Errorf("-s3 required (host:port of s3 server gRPC)")
+	}
+	if *eventBudget < 0 {
+		return fmt.Errorf("-events must be >= 0 (0 = unbounded)")
+	}
+
+	bucketsPath, err := resolveBucketsPath(env)
+	if err != nil {
+		return fmt.Errorf("resolve buckets path: %w", err)
+	}
+	fmt.Fprintf(writer, "buckets path: %s\n", bucketsPath)
+
+	dialCtx, dialCancel := context.WithTimeout(context.Background(), 30*time.Second)
+	conn, err := pb.GrpcDial(dialCtx, *s3Endpoint, false, env.option.GrpcDialOption)
+	dialCancel()
+	if err != nil {
+		return fmt.Errorf("dial s3 %s: %w", *s3Endpoint, err)
+	}
+	defer conn.Close()
+	rpcClient := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(conn)
+
+	// Run the whole pipeline inside one WithFilerClient so the reader's
+	// SubscribeMetadata stream and the persister share a single connection.
+	return env.WithFilerClient(true, func(filerClient filer_pb.SeaweedFilerClient) error {
+		inputs, parseErrors, err := loadLifecycleCompileInputs(context.Background(), filerClient, bucketsPath)
+		if err != nil {
+			return fmt.Errorf("load lifecycle configs: %w", err)
+		}
+		for i, pe := range parseErrors {
+			// Surface up to the first three parse errors so the operator
+			// can chase malformed configs; cap the rest with a count so
+			// the output stays readable on large clusters.
+			if i < 3 {
+				fmt.Fprintf(writer, "warning: %s: %v\n", pe.bucket, pe.err)
+			}
+		}
+		if extra := len(parseErrors) - 3; extra > 0 {
+			fmt.Fprintf(writer, "warning: %d additional bucket(s) had malformed lifecycle config\n", extra)
+		}
+		if len(inputs) == 0 {
+			fmt.Fprintln(writer, "no buckets with enabled lifecycle rules found")
+			return nil
+		}
+		fmt.Fprintf(writer, "loaded lifecycle for %d bucket(s)\n", len(inputs))
+
+		// Activate every action so this manual run dispatches whatever fires.
+		// The production bootstrap walker promotes actions only after a clean
+		// walk; this shell entrypoint runs out-of-band of that flow.
+		eng := engine.New()
+		eng.Compile(inputs, engine.CompileOptions{PriorStates: allActivePriorStates(inputs)})
+
+		pipeline := &dispatcher.Pipeline{
+			Shards:         shards,
+			BucketsPath:    bucketsPath,
+			Engine:         eng,
+			Persister:      &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
+			Client:         &lifecycleClientCallable{c: rpcClient},
+			FilerClient:    filerClient,
+			ClientID:       util.RandomInt32(),
+			ClientName:     fmt.Sprintf("shell-lifecycle-%s", formatShardLabel(shards)),
+			DispatchTick:   *dispatchTick,
+			CheckpointTick: *checkpointTick,
+			EventBudget:    *eventBudget,
+		}
+
+		var ctx context.Context
+		var cancel context.CancelFunc
+		if *runtime > 0 {
+			ctx, cancel = context.WithTimeout(context.Background(), *runtime)
+		} else {
+			ctx, cancel = context.WithCancel(context.Background())
+		}
+		defer cancel()
+
+		fmt.Fprintf(writer, "running shards %s (event budget=%d, runtime=%s)…\n", formatShardLabel(shards), *eventBudget, *runtime)
+		if err := pipeline.Run(ctx); err != nil {
+			return fmt.Errorf("pipeline: %w", err)
+		}
+		fmt.Fprintf(writer, "shards %s complete; cursors checkpointed\n", formatShardLabel(shards))
+		return nil
+	})
+}
+
+// resolveShardSelection turns the -shard / -shards flags into a sorted,
+// deduplicated []int. Exactly one form must be specified.
+func resolveShardSelection(singleShard int, shardsSpec string) ([]int, error) {
+	if singleShard >= 0 && shardsSpec != "" {
+		return nil, fmt.Errorf("-shard and -shards are mutually exclusive")
+	}
+	if singleShard < 0 && shardsSpec == "" {
+		return nil, fmt.Errorf("specify -shard <id> or -shards <range|set>")
+	}
+	if singleShard >= 0 {
+		if singleShard >= s3lifecycle.ShardCount {
+			return nil, fmt.Errorf("-shard %d out of [0,%d)", singleShard, s3lifecycle.ShardCount)
+		}
+		return []int{singleShard}, nil
+	}
+	return parseShardsSpec(shardsSpec)
+}
+
+// parseShardsSpec accepts "lo-hi" (inclusive) or "a,b,c" and returns a
+// sorted, deduplicated, in-range []int.
+func parseShardsSpec(spec string) ([]int, error) {
+	spec = strings.TrimSpace(spec)
+	seen := map[int]struct{}{}
+	add := func(v int) error {
+		if v < 0 || v >= s3lifecycle.ShardCount {
+			return fmt.Errorf("shard %d out of [0,%d)", v, s3lifecycle.ShardCount)
+		}
+		seen[v] = struct{}{}
+		return nil
+	}
+	if strings.Contains(spec, "-") && !strings.Contains(spec, ",") {
+		parts := strings.SplitN(spec, "-", 2)
+		lo, err := strconv.Atoi(strings.TrimSpace(parts[0]))
+		if err != nil {
+			return nil, fmt.Errorf("range lo: %w", err)
+		}
+		hi, err := strconv.Atoi(strings.TrimSpace(parts[1]))
+		if err != nil {
+			return nil, fmt.Errorf("range hi: %w", err)
+		}
+		if lo > hi {
+			return nil, fmt.Errorf("range lo %d > hi %d", lo, hi)
+		}
+		for v := lo; v <= hi; v++ {
+			if err := add(v); err != nil {
+				return nil, err
+			}
+		}
+	} else {
+		for _, part := range strings.Split(spec, ",") {
+			part = strings.TrimSpace(part)
+			if part == "" {
+				continue
+			}
+			v, err := strconv.Atoi(part)
+			if err != nil {
+				return nil, fmt.Errorf("shard list: %w", err)
+			}
+			if err := add(v); err != nil {
+				return nil, err
+			}
+		}
+	}
+	if len(seen) == 0 {
+		return nil, fmt.Errorf("empty shard set")
+	}
+	out := make([]int, 0, len(seen))
+	for v := range seen {
+		out = append(out, v)
+	}
+	sort.Ints(out)
+	return out, nil
+}
+
+func formatShardLabel(shards []int) string {
+	if len(shards) == 1 {
+		return fmt.Sprintf("%d", shards[0])
+	}
+	// Detect contiguous range.
+	contiguous := true
+	for i := 1; i < len(shards); i++ {
+		if shards[i] != shards[i-1]+1 {
+			contiguous = false
+			break
+		}
+	}
+	if contiguous {
+		return fmt.Sprintf("%d-%d", shards[0], shards[len(shards)-1])
+	}
+	parts := make([]string, len(shards))
+	for i, v := range shards {
+		parts[i] = strconv.Itoa(v)
+	}
+	return strings.Join(parts, ",")
+}
+
+// lifecycleClientCallable adapts the generated grpc client (variadic
+// CallOption tail) to the dispatcher.LifecycleClient interface.
+type lifecycleClientCallable struct {
+	c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient
+}
+
+func (l *lifecycleClientCallable) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+	return l.c.LifecycleDelete(ctx, req)
+}
+
+// resolveBucketsPath fetches the filer's configured buckets directory.
+// Falls back to /buckets when the filer doesn't return one.
+func resolveBucketsPath(env *CommandEnv) (string, error) {
+	var path string
+	err := env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		path = resp.GetDirBuckets()
+		return nil
+	})
+	if err != nil {
+		return "", err
+	}
+	if path == "" {
+		path = "/buckets"
+	}
+	return path, nil
+}
+
+type lifecycleParseError struct {
+	bucket string
+	err    error
+}
+
+// loadLifecycleCompileInputs walks the buckets directory and reads each
+// bucket entry's lifecycle XML from its Extended attributes. Pagination
+// loops with startFrom so clusters with more than one page of buckets
+// don't drop the tail. Parse errors are collected per bucket and returned
+// alongside the successful inputs so the caller can surface them.
+func loadLifecycleCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketsPath string) ([]engine.CompileInput, []lifecycleParseError, error) {
+	var (
+		inputs      []engine.CompileInput
+		parseErrors []lifecycleParseError
+		startFrom   string
+	)
+	const pageSize uint32 = 1024
+	for {
+		pageCount := 0
+		var lastName string
+		err := filer_pb.SeaweedList(ctx, client, bucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+			pageCount++
+			lastName = entry.Name
+			if !entry.IsDirectory {
+				return nil
+			}
+			xmlBytes, ok := entry.Extended[bucketLifecycleConfigurationXMLKey]
+			if !ok || len(xmlBytes) == 0 {
+				return nil
+			}
+			rules, err := lifecycle_xml.ParseCanonical(xmlBytes)
+			if err != nil {
+				parseErrors = append(parseErrors, lifecycleParseError{bucket: entry.Name, err: err})
+				return nil
+			}
+			if len(rules) == 0 {
+				return nil
+			}
+			inputs = append(inputs, engine.CompileInput{
+				Bucket:    entry.Name,
+				Rules:     rules,
+				Versioned: isBucketVersioned(entry),
+			})
+			return nil
+		}, startFrom, false, pageSize)
+		if err != nil {
+			return nil, nil, err
+		}
+		if uint32(pageCount) < pageSize {
+			break
+		}
+		startFrom = lastName
+	}
+	return inputs, parseErrors, nil
+}
+
+const bucketLifecycleConfigurationXMLKey = "s3-bucket-lifecycle-configuration-xml"
+
+func isBucketVersioned(entry *filer_pb.Entry) bool {
+	v, ok := entry.Extended[s3_constants.ExtVersioningKey]
+	if !ok {
+		return false
+	}
+	s := strings.ToLower(strings.TrimSpace(string(v)))
+	return s == "enabled" || s == "suspended"
+}
+
+// allActivePriorStates seeds every compiled action as bootstrap-complete +
+// event-driven so the run dispatches whatever fires. Production bootstrap
+// walks set this incrementally per bucket; this manual run skips the walk.
+func allActivePriorStates(inputs []engine.CompileInput) map[s3lifecycle.ActionKey]engine.PriorState {
+	prior := map[s3lifecycle.ActionKey]engine.PriorState{}
+	for _, in := range inputs {
+		for _, rule := range in.Rules {
+			hash := s3lifecycle.RuleHash(rule)
+			for _, kind := range s3lifecycle.RuleActionKinds(rule) {
+				key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: hash, ActionKind: kind}
+				prior[key] = engine.PriorState{
+					BootstrapComplete: true,
+					Mode:              engine.ModeEventDriven,
+				}
+			}
+		}
+	}
+	return prior
+}
diff --git a/weed/shell/command_s3_lifecycle_run_shard_test.go b/weed/shell/command_s3_lifecycle_run_shard_test.go
new file mode 100644
index 000000000..a1f5321ef
--- /dev/null
+++ b/weed/shell/command_s3_lifecycle_run_shard_test.go
@@ -0,0 +1,92 @@
+package shell
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestParseShardsSpec_Range(t *testing.T) {
+	got, err := parseShardsSpec("3-7")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	want := []int{3, 4, 5, 6, 7}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestParseShardsSpec_Set(t *testing.T) {
+	got, err := parseShardsSpec("0,3,7")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	want := []int{0, 3, 7}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestParseShardsSpec_DedupSort(t *testing.T) {
+	got, err := parseShardsSpec("7,3,3,0")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	want := []int{0, 3, 7}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestParseShardsSpec_OutOfRange(t *testing.T) {
+	if _, err := parseShardsSpec("16"); err == nil {
+		t.Fatal("expected out-of-range error for 16")
+	}
+	if _, err := parseShardsSpec("-1"); err == nil {
+		t.Fatal("expected out-of-range error for -1")
+	}
+	if _, err := parseShardsSpec("0-16"); err == nil {
+		t.Fatal("expected out-of-range error for 0-16")
+	}
+}
+
+func TestParseShardsSpec_BadRange(t *testing.T) {
+	if _, err := parseShardsSpec("7-3"); err == nil {
+		t.Fatal("expected lo>hi error")
+	}
+}
+
+func TestResolveShardSelection_Mutex(t *testing.T) {
+	if _, err := resolveShardSelection(0, "1,2"); err == nil {
+		t.Fatal("expected mutex error when both -shard and -shards set")
+	}
+	if _, err := resolveShardSelection(-1, ""); err == nil {
+		t.Fatal("expected error when neither set")
+	}
+}
+
+func TestResolveShardSelection_SingleShard(t *testing.T) {
+	got, err := resolveShardSelection(5, "")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if !reflect.DeepEqual(got, []int{5}) {
+		t.Fatalf("got %v, want [5]", got)
+	}
+}
+
+func TestFormatShardLabel(t *testing.T) {
+	cases := []struct {
+		in   []int
+		want string
+	}{
+		{[]int{5}, "5"},
+		{[]int{0, 1, 2, 3}, "0-3"},
+		{[]int{0, 2, 5}, "0,2,5"},
+	}
+	for _, tc := range cases {
+		if got := formatShardLabel(tc.in); got != tc.want {
+			t.Errorf("formatShardLabel(%v)=%q, want %q", tc.in, got, tc.want)
+		}
+	}
+}