diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go
index 4faca8ab8..5f3c1bfed 100644
--- a/weed/worker/tasks/iceberg/detection.go
+++ b/weed/worker/tasks/iceberg/detection.go
@@ -486,27 +486,12 @@ func compactionMinInputFiles(minInputFiles int64) (int, error) {
 
 // needsMaintenance checks whether snapshot expiration work is needed based on
 // metadata-only thresholds.
+// It asks execution what it would do rather than reimplementing the rules:
+// expiry always requires a snapshot to be past the retention window, so a
+// table over the snapshot quota whose snapshots are all young, or all pinned by
+// refs, would otherwise be proposed for a job that can only no-op.
 func needsMaintenance(meta table.Metadata, config Config) bool {
-	snapshots := meta.Snapshots()
-	if len(snapshots) == 0 {
-		return false
-	}
-
-	// Check snapshot count
-	if int64(len(snapshots)) > config.MaxSnapshotsToKeep {
-		return true
-	}
-
-	// Check oldest snapshot age
-	retentionMs := config.SnapshotRetentionMs
-	nowMs := time.Now().UnixMilli()
-	for _, snap := range snapshots {
-		if nowMs-snap.TimestampMs > retentionMs {
-			return true
-		}
-	}
-
-	return false
+	return len(snapshotsToExpire(meta, config, time.Now().UnixMilli())) > 0
 }
 
 // buildMaintenanceProposal creates a JobProposal for a table needing maintenance.
diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go
index 70033b4f4..cb2dd9eab 100644
--- a/weed/worker/tasks/iceberg/exec_test.go
+++ b/weed/worker/tasks/iceberg/exec_test.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -40,6 +41,7 @@ type fakeFilerServer struct {
 	mu           sync.Mutex
 	entries      map[string]map[string]*filer_pb.Entry // dir → name → entry
 	beforeUpdate func(*fakeFilerServer, *filer_pb.UpdateEntryRequest) error
+	beforeLookup func(*fakeFilerServer, *filer_pb.LookupDirectoryEntryRequest)
 
 	// Set by enableAssign to serve AssignVolume against a fake volume server.
 	assignVolumeServer string
@@ -93,6 +95,13 @@ func (f *fakeFilerServer) listDir(dir string) []*filer_pb.Entry {
 }
 
 func (f *fakeFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	f.mu.Lock()
+	beforeLookup := f.beforeLookup
+	f.mu.Unlock()
+	if beforeLookup != nil {
+		beforeLookup(f, req)
+	}
+
 	entry := f.getEntry(req.Directory, req.Name)
 	if entry == nil {
 		return nil, status.Errorf(codes.NotFound, "entry not found: %s/%s", req.Directory, req.Name)
@@ -287,6 +296,12 @@ type tableSetup struct {
 	// that created it would record the table's running totals.
 	SnapshotSummary map[string]string
 	Snapshots       []table.Snapshot
+	// Refs are branches and tags beyond main, which always points at the last
+	// snapshot.
+	Refs map[string]table.SnapshotRef
+	// Age backdates the whole table so its snapshots can sit outside a
+	// retention window.
+	Age time.Duration
 }
 
 func (ts tableSetup) tablePath() string {
@@ -314,7 +329,7 @@ func (ts tableSetup) fileRef(elem ...string) string {
 func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Metadata {
 	t.Helper()
 
-	meta := buildTestMetadata(t, setup.Snapshots)
+	meta := buildTestMetadata(t, setup.Snapshots, setup.Refs, setup.Age)
 	fullMetadataJSON, err := json.Marshal(meta)
 	if err != nil {
 		t.Fatalf("marshal metadata: %v", err)
@@ -582,6 +597,244 @@ func TestExpireSnapshotsExecution(t *testing.T) {
 	}
 }
 
+func TestExpireSnapshotsKeepsTaggedSnapshot(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().Add(-10 * time.Second).UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+		},
+		Refs: map[string]table.SnapshotRef{
+			"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
+		},
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
+	}
+
+	if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("reload metadata: %v", err)
+	}
+
+	remaining := map[int64]bool{}
+	for _, snap := range state.Metadata.Snapshots() {
+		remaining[snap.SnapshotID] = true
+	}
+	if !remaining[1] {
+		t.Error("tagged snapshot 1 was expired")
+	}
+	if remaining[2] {
+		t.Error("expected untagged snapshot 2 to be expired")
+	}
+
+	var tagged bool
+	for name, ref := range state.Metadata.Refs() {
+		if name == "release" {
+			tagged = true
+			if ref.SnapshotID != 1 {
+				t.Errorf("tag release points at snapshot %d, want 1", ref.SnapshotID)
+			}
+		}
+	}
+	if !tagged {
+		t.Error("tag release was dropped from the metadata")
+	}
+
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
+	if fs.getEntry(metaDir, "snap-1.avro") == nil {
+		t.Error("manifest list of the tagged snapshot was deleted")
+	}
+	if fs.getEntry(metaDir, "snap-2.avro") != nil {
+		t.Error("expected the expired snapshot's manifest list to be deleted")
+	}
+}
+
+func TestExpireSnapshotsHonorsBranchRetention(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	// The branch chain is 1 <- 2 <- 3, with main on 4. Backdating the table by
+	// Age gives every snapshot a real age; the ten-second spacing keeps them
+	// distinct and stays inside what iceberg-go accepts at build time.
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	stepMs := int64(10 * time.Second / time.Millisecond)
+	first, second := int64(1), int64(2)
+	minKeep := 2
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
+			{SnapshotID: 2, TimestampMs: now + stepMs, ManifestList: "metadata/snap-2.avro", ParentSnapshotID: &first, SequenceNumber: 2},
+			{SnapshotID: 3, TimestampMs: now + 2*stepMs, ManifestList: "metadata/snap-3.avro", ParentSnapshotID: &second, SequenceNumber: 3},
+			{SnapshotID: 4, TimestampMs: now + 3*stepMs, ManifestList: "metadata/snap-4.avro", SequenceNumber: 4},
+		},
+		Refs: map[string]table.SnapshotRef{
+			"audit": {SnapshotID: 3, SnapshotRefType: table.BranchRef, MinSnapshotsToKeep: &minKeep},
+		},
+		Age: 5 * time.Hour,
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
+	}
+
+	if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+
+	// min-snapshots-to-keep=2 reaches from the branch head back to its parent,
+	// and stops there: the grandparent is past the count and expires.
+	assertSnapshots(t, client, setup, []int64{2, 3, 4}, []int64{1})
+}
+
+func TestExpireSnapshotsHonorsBranchMaxSnapshotAge(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	stepMs := int64(10 * time.Second / time.Millisecond)
+	first, second := int64(1), int64(2)
+	// Between the branch head's parent (5h20s) and its grandparent (5h30s), so
+	// the window has to reach past the head to be observable.
+	maxAgeMs := int64(5*time.Hour/time.Millisecond) + 25*1000
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
+			{SnapshotID: 2, TimestampMs: now + stepMs, ManifestList: "metadata/snap-2.avro", ParentSnapshotID: &first, SequenceNumber: 2},
+			{SnapshotID: 3, TimestampMs: now + 2*stepMs, ManifestList: "metadata/snap-3.avro", ParentSnapshotID: &second, SequenceNumber: 3},
+			{SnapshotID: 4, TimestampMs: now + 3*stepMs, ManifestList: "metadata/snap-4.avro", SequenceNumber: 4},
+		},
+		Refs: map[string]table.SnapshotRef{
+			"audit": {SnapshotID: 3, SnapshotRefType: table.BranchRef, MaxSnapshotAgeMs: &maxAgeMs},
+		},
+		Age: 5 * time.Hour,
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
+	}
+
+	if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+
+	// The window reaches from the branch head back over its parent but stops
+	// short of the grandparent.
+	assertSnapshots(t, client, setup, []int64{2, 3, 4}, []int64{1})
+}
+
+// assertSnapshots reloads the table and checks exactly which snapshots survived.
+func assertSnapshots(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, want, gone []int64) {
+	t.Helper()
+
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("reload metadata: %v", err)
+	}
+	remaining := map[int64]bool{}
+	for _, snap := range state.Metadata.Snapshots() {
+		remaining[snap.SnapshotID] = true
+	}
+	for _, id := range want {
+		if !remaining[id] {
+			t.Errorf("snapshot %d was expired, want kept", id)
+		}
+	}
+	for _, id := range gone {
+		if remaining[id] {
+			t.Errorf("snapshot %d survived, want expired", id)
+		}
+	}
+}
+
+// A tag created between planning and commit pins a snapshot the plan was going
+// to expire. The head has not moved, so only a refs check can catch it.
+func TestExpireSnapshotsRejectsPlanWhenARefAppears(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+		},
+		Age: 200 * time.Hour,
+	}
+	populateTable(t, fs, setup)
+
+	// Re-publish the table with a tag on snapshot 1 while the commit is
+	// re-reading it, which is the window the guard closes.
+	tagged := setup
+	tagged.Refs = map[string]table.SnapshotRef{
+		"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
+	}
+	lookups := 0
+	fs.beforeLookup = func(f *fakeFilerServer, req *filer_pb.LookupDirectoryEntryRequest) {
+		if req.Name != setup.TableName {
+			return
+		}
+		lookups++
+		if lookups == 2 {
+			populateTable(t, f, tagged)
+		}
+	}
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    1,
+		Operations:          "expire_snapshots",
+	}
+
+	_, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err == nil {
+		t.Fatal("expireSnapshots() error = nil, want the plan rejected as stale")
+	}
+	if !errors.Is(err, errStalePlan) {
+		t.Fatalf("expireSnapshots() error = %v, want errStalePlan", err)
+	}
+
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
+	if fs.getEntry(metaDir, "snap-1.avro") == nil {
+		t.Error("the newly tagged snapshot's manifest list was deleted")
+	}
+}
+
 func TestExpireSnapshotsNothingToExpire(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -1239,6 +1492,7 @@ func TestDetectWithFakeFiler(t *testing.T) {
 			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
 			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
 		},
+		Age: 200 * time.Hour, // past the default 7-day retention
 	}
 	populateTable(t, fs, setup)
 
@@ -1286,6 +1540,7 @@ func TestDetectWithFilters(t *testing.T) {
 			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
 			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
 		},
+		Age: 200 * time.Hour,
 	}
 	setup2 := tableSetup{
 		BucketName: "bucket-b",
@@ -1296,6 +1551,7 @@ func TestDetectWithFilters(t *testing.T) {
 			{SnapshotID: 5, TimestampMs: now + 4, ManifestList: "metadata/snap-5.avro"},
 			{SnapshotID: 6, TimestampMs: now + 5, ManifestList: "metadata/snap-6.avro"},
 		},
+		Age: 200 * time.Hour,
 	}
 	populateTable(t, fs, setup1)
 	populateTable(t, fs, setup2)
@@ -1523,6 +1779,7 @@ func TestDetectSchedulesSnapshotExpiryDespiteCompactionEvaluationError(t *testin
 			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
 			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro", SequenceNumber: 2},
 		},
+		Age: 400 * 24 * time.Hour, // past the year-long retention below
 	}
 	populateTable(t, fs, setup)
 
diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go
index 8a3bf7f23..f2b4818fe 100644
--- a/weed/worker/tasks/iceberg/handler_test.go
+++ b/weed/worker/tasks/iceberg/handler_test.go
@@ -3,9 +3,11 @@ package iceberg
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"path"
+	"strconv"
 	"testing"
 	"time"
 
@@ -119,13 +121,33 @@ func TestNeedsMaintenanceNoSnapshots(t *testing.T) {
 		MaxSnapshotsToKeep:  2,
 	}
 
-	meta := buildTestMetadata(t, nil)
+	meta := buildTestMetadata(t, nil, nil, 0)
 	if needsMaintenance(meta, config) {
 		t.Error("expected no maintenance for table with no snapshots")
 	}
 }
 
 func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(24),
+		MaxSnapshotsToKeep:  2,
+	}
+
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+		{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+	}
+	meta := buildTestMetadata(t, snapshots, nil, 48*time.Hour)
+	if !needsMaintenance(meta, config) {
+		t.Error("expected maintenance for table exceeding max snapshots")
+	}
+}
+
+// Expiry always requires a snapshot to be past the retention window, so a table
+// over the quota whose snapshots are all young has nothing to do.
+func TestNeedsMaintenanceExceedsMaxSnapshotsWithinRetention(t *testing.T) {
 	config := Config{
 		SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention
 		MaxSnapshotsToKeep:  2,
@@ -137,9 +159,31 @@ func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
 		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
 		{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
 	}
-	meta := buildTestMetadata(t, snapshots)
-	if !needsMaintenance(meta, config) {
-		t.Error("expected maintenance for table exceeding max snapshots")
+	if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0), config) {
+		t.Error("expected no maintenance while every snapshot is inside the retention window")
+	}
+}
+
+// Every expirable snapshot is pinned by a tag, so a job could only no-op.
+func TestNeedsMaintenanceSkipsRefPinnedSnapshots(t *testing.T) {
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+	}
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+	}
+	refs := map[string]table.SnapshotRef{
+		"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
+	}
+	if needsMaintenance(buildTestMetadata(t, snapshots, refs, 0), config) {
+		t.Error("expected no maintenance when the only old snapshot is tagged")
+	}
+	if !needsMaintenance(buildTestMetadata(t, snapshots, nil, 0), config) {
+		t.Error("expected maintenance for the same table without the tag")
 	}
 }
 
@@ -153,7 +197,7 @@ func TestNeedsMaintenanceWithinLimits(t *testing.T) {
 	snapshots := []table.Snapshot{
 		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
 	}
-	meta := buildTestMetadata(t, snapshots)
+	meta := buildTestMetadata(t, snapshots, nil, 0)
 	if needsMaintenance(meta, config) {
 		t.Error("expected no maintenance for table within limits")
 	}
@@ -163,17 +207,34 @@ func TestNeedsMaintenanceOldSnapshot(t *testing.T) {
 	// Use a retention of 0 hours so that any snapshot is considered "old"
 	config := Config{
 		SnapshotRetentionMs: hoursToMs(0), // instant expiry
+		MaxSnapshotsToKeep:  1,
+	}
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+	}
+	meta := buildTestMetadata(t, snapshots, nil, 0)
+	if !needsMaintenance(meta, config) {
+		t.Error("expected maintenance for table with expired snapshot")
+	}
+}
+
+// The current snapshot is never expirable, so a single-snapshot table has no
+// work no matter how far past retention it is.
+func TestNeedsMaintenanceSingleSnapshot(t *testing.T) {
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
 		MaxSnapshotsToKeep:  10,
 	}
 
-	now := time.Now().UnixMilli()
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
 	snapshots := []table.Snapshot{
-		{SnapshotID: 1, TimestampMs: now - 1, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
 	}
-	meta := buildTestMetadata(t, snapshots)
-	// With 0 retention, any snapshot with timestamp < now should need maintenance
-	if !needsMaintenance(meta, config) {
-		t.Error("expected maintenance for table with expired snapshot")
+	if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0), config) {
+		t.Error("expected no maintenance for a table with only the current snapshot")
 	}
 }
 
@@ -217,7 +278,7 @@ func TestBuildMaintenanceProposal(t *testing.T) {
 		{SnapshotID: 1, TimestampMs: now},
 		{SnapshotID: 2, TimestampMs: now + 1},
 	}
-	meta := buildTestMetadata(t, snapshots)
+	meta := buildTestMetadata(t, snapshots, nil, 0)
 
 	info := tableInfo{
 		BucketName: "my-bucket",
@@ -1408,9 +1469,14 @@ func TestExecuteNilRequest(t *testing.T) {
 // Test helpers
 // ---------------------------------------------------------------------------
 
-// buildTestMetadata creates a minimal Iceberg metadata for testing.
-// When snapshots is nil or empty, the metadata has no snapshots.
-func buildTestMetadata(t *testing.T, snapshots []table.Snapshot) table.Metadata {
+// buildTestMetadata creates a minimal Iceberg metadata for testing. When
+// snapshots is nil or empty the metadata has no snapshots; refs adds branches
+// and tags on top of the main branch, which always points at the last snapshot.
+// A positive age backdates the result, letting a test describe snapshots that
+// are genuinely past a retention window - iceberg-go refuses to add a snapshot
+// stamped more than a minute before the metadata's last-updated time, so the
+// shift has to happen after the build.
+func buildTestMetadata(t *testing.T, snapshots []table.Snapshot, refs map[string]table.SnapshotRef, age time.Duration) table.Metadata {
 	t.Helper()
 
 	schema := newTestSchema()
@@ -1441,11 +1507,80 @@ func buildTestMetadata(t *testing.T, snapshots []table.Snapshot) table.Metadata
 		t.Fatalf("failed to set snapshot ref: %v", err)
 	}
 
+	// The option type is unexported, so each combination is spelled out.
+	for name, ref := range refs {
+		var refErr error
+		switch {
+		case ref.MinSnapshotsToKeep != nil && ref.MaxSnapshotAgeMs != nil:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType,
+				table.WithMinSnapshotsToKeep(*ref.MinSnapshotsToKeep), table.WithMaxSnapshotAgeMs(*ref.MaxSnapshotAgeMs))
+		case ref.MinSnapshotsToKeep != nil:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType,
+				table.WithMinSnapshotsToKeep(*ref.MinSnapshotsToKeep))
+		case ref.MaxSnapshotAgeMs != nil:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType,
+				table.WithMaxSnapshotAgeMs(*ref.MaxSnapshotAgeMs))
+		default:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType)
+		}
+		if refErr != nil {
+			t.Fatalf("failed to set ref %s: %v", name, refErr)
+		}
+	}
+
 	result, err := builder.Build()
 	if err != nil {
 		t.Fatalf("failed to build metadata: %v", err)
 	}
-	return result
+	metadata := result
+	if age <= 0 {
+		return metadata
+	}
+
+	encoded, err := json.Marshal(metadata)
+	if err != nil {
+		t.Fatalf("marshal metadata: %v", err)
+	}
+
+	decoder := json.NewDecoder(bytes.NewReader(encoded))
+	decoder.UseNumber()
+	var doc map[string]any
+	if err := decoder.Decode(&doc); err != nil {
+		t.Fatalf("decode metadata: %v", err)
+	}
+
+	shiftMs := age.Milliseconds()
+	shift := func(container map[string]any, key string) {
+		raw, ok := container[key].(json.Number)
+		if !ok {
+			return
+		}
+		value, err := raw.Int64()
+		if err != nil {
+			t.Fatalf("parse %s: %v", key, err)
+		}
+		container[key] = json.Number(strconv.FormatInt(value-shiftMs, 10))
+	}
+
+	shift(doc, "last-updated-ms")
+	for _, listKey := range []string{"snapshots", "snapshot-log", "metadata-log"} {
+		entries, _ := doc[listKey].([]any)
+		for _, entry := range entries {
+			if item, ok := entry.(map[string]any); ok {
+				shift(item, "timestamp-ms")
+			}
+		}
+	}
+
+	aged, err := json.Marshal(doc)
+	if err != nil {
+		t.Fatalf("marshal aged metadata: %v", err)
+	}
+	parsed, err := table.ParseMetadataBytes(aged)
+	if err != nil {
+		t.Fatalf("parse aged metadata: %v", err)
+	}
+	return parsed
 }
 
 func newTestSchema() *iceberg.Schema {
diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go
index d00202c22..d4355e219 100644
--- a/weed/worker/tasks/iceberg/operations.go
+++ b/weed/worker/tasks/iceberg/operations.go
@@ -51,47 +51,13 @@ func (h *Handler) expireSnapshots(
 		return "no snapshots", nil, nil
 	}
 
-	// Determine which snapshots to expire
 	currentSnap := meta.CurrentSnapshot()
 	var currentSnapID int64
 	if currentSnap != nil {
 		currentSnapID = currentSnap.SnapshotID
 	}
 
-	retentionMs := config.SnapshotRetentionMs
-	nowMs := time.Now().UnixMilli()
-
-	// Sort snapshots by timestamp descending (most recent first) so that
-	// the keep-count logic always preserves the newest snapshots.
-	sorted := make([]table.Snapshot, len(snapshots))
-	copy(sorted, snapshots)
-	sort.Slice(sorted, func(i, j int) bool {
-		return sorted[i].TimestampMs > sorted[j].TimestampMs
-	})
-
-	// Walk from newest to oldest. The current snapshot is always kept.
-	// Among the remaining, keep up to MaxSnapshotsToKeep-1 (since current
-	// counts toward the quota). Expire the rest only if they exceed the
-	// retention window; snapshots within the window are kept regardless.
-	var toExpire []int64
-	var kept int64
-	for _, snap := range sorted {
-		if snap.SnapshotID == currentSnapID {
-			kept++
-			continue
-		}
-		age := nowMs - snap.TimestampMs
-		if kept < config.MaxSnapshotsToKeep {
-			kept++
-			continue
-		}
-		if age > retentionMs {
-			toExpire = append(toExpire, snap.SnapshotID)
-		} else {
-			kept++
-		}
-	}
-
+	toExpire := snapshotsToExpire(meta, config, time.Now().UnixMilli())
 	if len(toExpire) == 0 {
 		return "no snapshots expired", nil, nil
 	}
@@ -102,7 +68,7 @@ func (h *Handler) expireSnapshots(
 		expireSet[id] = struct{}{}
 	}
 	var expiredSnaps, keptSnaps []table.Snapshot
-	for _, snap := range sorted {
+	for _, snap := range snapshots {
 		if _, ok := expireSet[snap.SnapshotID]; ok {
 			expiredSnaps = append(expiredSnaps, snap)
 		} else {
@@ -136,6 +102,15 @@ func (h *Handler) expireSnapshots(
 		if (cs == nil) != (currentSnapID == 0) || (cs != nil && cs.SnapshotID != currentSnapID) {
 			return errStalePlan
 		}
+		// A tag or branch created since planning can pin a snapshot this plan
+		// expires without moving the head, and RemoveSnapshots would drop that
+		// ref along with it. Re-plan instead.
+		nowProtected := protectedSnapshots(currentMeta, time.Now().UnixMilli())
+		for _, id := range toExpire {
+			if _, pinned := nowProtected[id]; pinned {
+				return errStalePlan
+			}
+		}
 		return builder.RemoveSnapshots(toExpire, false)
 	})
 	if err != nil {
@@ -168,6 +143,106 @@ func (h *Handler) expireSnapshots(
 	return fmt.Sprintf("expired %d snapshot(s), deleted %d unreferenced file(s)", len(toExpire), deletedCount), metrics, nil
 }
 
+// snapshotsToExpire returns the snapshot IDs that fall outside the configured
+// retention. Detection and execution both call it so a table is only proposed
+// for expiry when the run would actually remove something.
+func snapshotsToExpire(meta table.Metadata, config Config, nowMs int64) []int64 {
+	var currentSnapID int64
+	if currentSnap := meta.CurrentSnapshot(); currentSnap != nil {
+		currentSnapID = currentSnap.SnapshotID
+	}
+	retentionMs := config.SnapshotRetentionMs
+	protected := protectedSnapshots(meta, nowMs)
+
+	// Sort snapshots by timestamp descending (most recent first) so that
+	// the keep-count logic always preserves the newest snapshots.
+	sorted := make([]table.Snapshot, len(meta.Snapshots()))
+	copy(sorted, meta.Snapshots())
+	sort.Slice(sorted, func(i, j int) bool {
+		return sorted[i].TimestampMs > sorted[j].TimestampMs
+	})
+
+	// Walk from newest to oldest. The current snapshot is always kept.
+	// Among the remaining, keep up to MaxSnapshotsToKeep-1 (since current
+	// counts toward the quota). Expire the rest only if they exceed the
+	// retention window; snapshots within the window are kept regardless.
+	var toExpire []int64
+	var kept int64
+	for _, snap := range sorted {
+		if snap.SnapshotID == currentSnapID {
+			kept++
+			continue
+		}
+		if _, isProtected := protected[snap.SnapshotID]; isProtected {
+			kept++
+			continue
+		}
+		if kept < config.MaxSnapshotsToKeep {
+			kept++
+			continue
+		}
+		if nowMs-snap.TimestampMs > retentionMs {
+			toExpire = append(toExpire, snap.SnapshotID)
+		} else {
+			kept++
+		}
+	}
+	return toExpire
+}
+
+// protectedSnapshots returns the snapshots that named refs hold in place.
+// A branch head or a tag pins its snapshot no matter how old it is, and a
+// branch may carry its own retention overrides for the ancestors behind it.
+// iceberg-go's RemoveSnapshots drops any ref whose snapshot is gone without
+// complaint, so expiring one of these would silently delete the tag and then
+// the files it pointed at.
+func protectedSnapshots(meta table.Metadata, nowMs int64) map[int64]struct{} {
+	protected := make(map[int64]struct{})
+	byID := make(map[int64]table.Snapshot)
+	for _, snap := range meta.Snapshots() {
+		byID[snap.SnapshotID] = snap
+	}
+
+	for _, ref := range meta.Refs() {
+		protected[ref.SnapshotID] = struct{}{}
+		if ref.SnapshotRefType != table.BranchRef {
+			continue
+		}
+		// Without overrides the branch keeps only its head here; the ancestors
+		// behind it stay under the worker's own retention config.
+		if ref.MinSnapshotsToKeep == nil && ref.MaxSnapshotAgeMs == nil {
+			continue
+		}
+		minToKeep := 1
+		if ref.MinSnapshotsToKeep != nil {
+			minToKeep = *ref.MinSnapshotsToKeep
+		}
+		seen := make(map[int64]struct{})
+		kept := 0
+		for id := ref.SnapshotID; ; {
+			if _, looped := seen[id]; looped {
+				break
+			}
+			seen[id] = struct{}{}
+			snap, ok := byID[id]
+			if !ok {
+				break
+			}
+			withinAge := ref.MaxSnapshotAgeMs != nil && nowMs-snap.TimestampMs <= *ref.MaxSnapshotAgeMs
+			if kept >= minToKeep && !withinAge {
+				break
+			}
+			protected[snap.SnapshotID] = struct{}{}
+			kept++
+			if snap.ParentSnapshotID == nil {
+				break
+			}
+			id = *snap.ParentSnapshotID
+		}
+	}
+	return protected
+}
+
 // collectSnapshotFiles returns all file paths (manifest lists, manifest files,
 // data files) referenced by the given snapshots. It returns an error if any
 // manifest list or manifest cannot be read/parsed, to prevent delete decisions