From bfd0cbccecdeebc7304a93d51493ae73fc1be788 Mon Sep 17 00:00:00 2001
From: Luiz Oliveira <luizaoj@google.com>
Date: Wed, 16 Sep 2026 16:01:31 -0400
Subject: [PATCH 1/6] Index the Actors borrowing a Tag's snapshot

---
 cmd/ateapi/internal/controlapi/tag.go         |   4 +
 cmd/ateapi/internal/store/atepg/atepg.go      | 110 ++++++-
 cmd/ateapi/internal/store/atepg/atepg_test.go |   2 +-
 .../store/atepg/migrations/000001_initial.sql |   9 +
 cmd/ateapi/internal/store/atepg/pagetoken.go  |   1 +
 cmd/ateapi/internal/store/store.go            |   4 +
 .../internal/store/storecontract/contract.go  | 299 ++++++++++++++++++
 internal/resources/snapshot_test.go           |  35 ++
 8 files changed, 461 insertions(+), 3 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/tag.go b/cmd/ateapi/internal/controlapi/tag.go
index 13037e1602..a5ced8e6dd 100644
--- a/cmd/ateapi/internal/controlapi/tag.go
+++ b/cmd/ateapi/internal/controlapi/tag.go
@@ -317,6 +317,10 @@ func (s *ServiceImpl) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*
 	return s.store.DeleteTag(ctx, tagRef)
 }
 
+func (s *ServiceImpl) ListTagBorrowers(ctx context.Context, tagUID string, opts store.ListOptions) (store.ListResponse[string], error) {
+	return s.store.ListTagBorrowers(ctx, tagUID, opts)
+}
+
 func validateDeleteTagRequest(ctx context.Context, req *ateapipb.DeleteTagRequest) field.ErrorList {
 	op := operation.Operation{Type: operation.Create}
 	return Validate_DeleteTagRequest(ctx, op, nil, req, nil)
diff --git a/cmd/ateapi/internal/store/atepg/atepg.go b/cmd/ateapi/internal/store/atepg/atepg.go
index c383a6bef5..86f0b81c9b 100644
--- a/cmd/ateapi/internal/store/atepg/atepg.go
+++ b/cmd/ateapi/internal/store/atepg/atepg.go
@@ -681,7 +681,13 @@ func (p *Persistence) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*
 		return nil, fmt.Errorf("marshaling actor: %w", err)
 	}
 
-	_, err = p.pool.Exec(ctx, `
+	tx, err := p.pool.Begin(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("beginning actor create: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
+
+	_, err = tx.Exec(ctx, `
 		INSERT INTO actors (atespace, name, uid, version, proto)
 		VALUES ($1, $2, $3, $4, $5)`,
 		atespace, name, dbActor.GetMetadata().GetUid(), dbActor.GetMetadata().GetVersion(), protoBytes)
@@ -696,9 +702,47 @@ func (p *Persistence) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*
 		}
 		return nil, fmt.Errorf("inserting actor %s/%s: %w", atespace, name, err)
 	}
+	if err := updateTagBorrow(ctx, tx, dbActor); err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, fmt.Errorf("committing actor create: %w", err)
+	}
 	return dbActor, nil
 }
 
+// updateTagBorrow upserts or clears the actor's borrow of a Tag's external
+// snapshot.
+func updateTagBorrow(ctx context.Context, tx pgx.Tx, actor *ateapipb.Actor) error {
+	actorUID := actor.GetMetadata().GetUid()
+
+	var tagUID string
+	if snapshotURI := actor.GetStatus().GetExternalSnapshot().GetSnapshotUri(); snapshotURI != "" {
+		uri, err := resources.ParseSnapshotURI(snapshotURI)
+		if err != nil {
+			return fmt.Errorf("reading the external snapshot of actor %s: %w", actorUID, err)
+		}
+		if owner, ok := uri.Owner().TagUID(); ok {
+			tagUID = owner
+		}
+	}
+
+	if tagUID == "" {
+		if _, err := tx.Exec(ctx, `DELETE FROM tag_borrows WHERE actor_uid = $1`, actorUID); err != nil {
+			return fmt.Errorf("clearing the tag borrow of actor %s: %w", actorUID, err)
+		}
+		return nil
+	}
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO tag_borrows (actor_uid, tag_uid)
+		VALUES ($1, $2)
+		ON CONFLICT (actor_uid) DO UPDATE SET tag_uid = $2`,
+		actorUID, tagUID); err != nil {
+		return fmt.Errorf("recording the borrow of tag %s by actor %s: %w", tagUID, actorUID, err)
+	}
+	return nil
+}
+
 func (p *Persistence) GetActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error) {
 	var protoBytes []byte
 	err := p.pool.QueryRow(ctx, `SELECT proto FROM actors WHERE atespace = $1 AND name = $2`, actorRef.Atespace, actorRef.Name).Scan(&protoBytes)
@@ -754,7 +798,13 @@ func (p *Persistence) UpdateActor(ctx context.Context, actorRef resources.ActorR
 	if err != nil {
 		return nil, fmt.Errorf("marshaling actor: %w", err)
 	}
-	commandTag, err := p.pool.Exec(ctx, `
+	tx, err := p.pool.Begin(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("beginning actor update: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
+
+	commandTag, err := tx.Exec(ctx, `
 			UPDATE actors
 			SET version = $1, proto = $2
 			WHERE atespace = $3 AND name = $4 AND uid = $5 AND version = $6`,
@@ -768,6 +818,12 @@ func (p *Persistence) UpdateActor(ctx context.Context, actorRef resources.ActorR
 	if commandTag.RowsAffected() != 1 {
 		return nil, fmt.Errorf("updating actor %s/%s affected %d rows, want 1", atespace, name, commandTag.RowsAffected())
 	}
+	if err := updateTagBorrow(ctx, tx, dbActor); err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, fmt.Errorf("committing actor update: %w", err)
+	}
 	return dbActor, nil
 }
 
@@ -803,6 +859,9 @@ func (p *Persistence) DeleteActor(ctx context.Context, actorRef resources.ActorR
 	if _, err := tx.Exec(ctx, `DELETE FROM actors WHERE atespace = $1 AND name = $2`, atespace, name); err != nil {
 		return nil, fmt.Errorf("deleting actor %s/%s: %w", atespace, name, err)
 	}
+	if _, err := tx.Exec(ctx, `DELETE FROM tag_borrows WHERE actor_uid = $1`, out.GetMetadata().GetUid()); err != nil {
+		return nil, fmt.Errorf("clearing the tag borrow of actor %s/%s: %w", atespace, name, err)
+	}
 	if err := tx.Commit(ctx); err != nil {
 		return nil, fmt.Errorf("committing actor delete: %w", err)
 	}
@@ -1296,6 +1355,53 @@ func (p *Persistence) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*
 	return tag, nil
 }
 
+func (p *Persistence) ListTagBorrowers(ctx context.Context, tagUID string, opts store.ListOptions) (store.ListResponse[string], error) {
+	opts, err := store.NormalizeListOptions(opts)
+	if err != nil {
+		return store.ListResponse[string]{}, err
+	}
+	pageSize := opts.PageSize
+	// The token is scoped to the Tag, so one cannot be replayed against another
+	// Tag's borrowers.
+	token, err := decodePageToken(opts.PageToken, kindTagBorrow, tagUID, 1)
+	if err != nil {
+		return store.ListResponse[string]{}, err
+	}
+	var last *string
+	if len(token.Last) > 0 {
+		last = &token.Last[0]
+	}
+
+	rows, err := p.pool.Query(ctx, `
+		SELECT actor_uid FROM tag_borrows
+		WHERE tag_uid = $1 AND ($2::text IS NULL OR actor_uid > $2)
+		ORDER BY actor_uid
+		LIMIT $3`, tagUID, last, int64(pageSize)+1)
+	if err != nil {
+		return store.ListResponse[string]{}, fmt.Errorf("listing the borrowers of tag %s: %w", tagUID, err)
+	}
+	defer rows.Close()
+
+	var actorUIDs []string
+	for rows.Next() {
+		var actorUID string
+		if err := rows.Scan(&actorUID); err != nil {
+			return store.ListResponse[string]{}, fmt.Errorf("scanning tag borrow row: %w", err)
+		}
+		actorUIDs = append(actorUIDs, actorUID)
+	}
+	if err := rows.Err(); err != nil {
+		return store.ListResponse[string]{}, fmt.Errorf("listing the borrowers of tag %s: %w", tagUID, err)
+	}
+
+	var nextToken string
+	if len(actorUIDs) > int(pageSize) {
+		actorUIDs = actorUIDs[:pageSize]
+		nextToken = encodePageToken(kindTagBorrow, tagUID, []string{actorUIDs[pageSize-1]})
+	}
+	return store.ListResponse[string]{Items: actorUIDs, NextPageToken: nextToken}, nil
+}
+
 // --- Workers ---
 
 func (p *Persistence) CreateWorker(ctx context.Context, worker *ateapipb.Worker) (*ateapipb.Worker, error) {
diff --git a/cmd/ateapi/internal/store/atepg/atepg_test.go b/cmd/ateapi/internal/store/atepg/atepg_test.go
index d2ed623991..fbb6ddcc6f 100644
--- a/cmd/ateapi/internal/store/atepg/atepg_test.go
+++ b/cmd/ateapi/internal/store/atepg/atepg_test.go
@@ -464,7 +464,7 @@ func appliedMigrationVersions(t *testing.T, pool *pgxpool.Pool) []int64 {
 // state, so the statement lives here rather than on Persistence.
 func clearAll(t *testing.T, p *Persistence) {
 	t.Helper()
-	if _, err := p.pool.Exec(context.Background(), `TRUNCATE atespaces, actors, actor_egress_policies, actor_templates, tags, workers, worker_assignments, leases, worker_outbox, worker_outbox_trim`); err != nil {
+	if _, err := p.pool.Exec(context.Background(), `TRUNCATE atespaces, actors, actor_egress_policies, actor_templates, tags, tag_borrows, workers, worker_assignments, leases, worker_outbox, worker_outbox_trim`); err != nil {
 		t.Fatalf("truncating tables: %v", err)
 	}
 }
diff --git a/cmd/ateapi/internal/store/atepg/migrations/000001_initial.sql b/cmd/ateapi/internal/store/atepg/migrations/000001_initial.sql
index 66ebbee243..664b5c88a2 100644
--- a/cmd/ateapi/internal/store/atepg/migrations/000001_initial.sql
+++ b/cmd/ateapi/internal/store/atepg/migrations/000001_initial.sql
@@ -86,6 +86,15 @@ CREATE TABLE worker_assignments (
 CREATE INDEX worker_assignments_worker_idx
     ON worker_assignments (worker_name);
 
+-- One row per Actor that borrows a Tag's external snapshot. Updated
+-- on every actor write.
+CREATE TABLE tag_borrows (
+    actor_uid  text PRIMARY KEY,
+    tag_uid    text NOT NULL
+);
+
+CREATE INDEX tag_borrows_tag_idx ON tag_borrows (tag_uid);
+
 -- Transactional outbox backing WatchWorkers.
 --
 -- 1. Ordering (xid): writeAndAppendEvent guarantees exactly one row per tx,
diff --git a/cmd/ateapi/internal/store/atepg/pagetoken.go b/cmd/ateapi/internal/store/atepg/pagetoken.go
index ebf56b5ae2..b03c1c9449 100644
--- a/cmd/ateapi/internal/store/atepg/pagetoken.go
+++ b/cmd/ateapi/internal/store/atepg/pagetoken.go
@@ -35,6 +35,7 @@ const (
 	kindActor         resourceKind = "actor"
 	kindActorTemplate resourceKind = "actor-template"
 	kindTag           resourceKind = "tag"
+	kindTagBorrow     resourceKind = "tag-borrow"
 	kindWorker        resourceKind = "worker"
 	kindWorkerAssign  resourceKind = "worker-assignment"
 )
diff --git a/cmd/ateapi/internal/store/store.go b/cmd/ateapi/internal/store/store.go
index 2d4b0522bd..5f53527876 100644
--- a/cmd/ateapi/internal/store/store.go
+++ b/cmd/ateapi/internal/store/store.go
@@ -154,6 +154,10 @@ type Interface interface {
 	// Deletes and returns a tag.
 	DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error)
 
+	// ListTagBorrowers returns a page of the UIDs of the Actors recorded as
+	// borrowing the external snapshot of the Tag with tagUID, ordered by UID.
+	ListTagBorrowers(ctx context.Context, tagUID string, opts ListOptions) (ListResponse[string], error)
+
 	// Stores a new atespace and returns the stored resource with server-assigned
 	// metadata (uid, version, timestamps). The input is not mutated. Returns
 	// ErrAlreadyExists if the name is taken.
diff --git a/cmd/ateapi/internal/store/storecontract/contract.go b/cmd/ateapi/internal/store/storecontract/contract.go
index 3514aece7e..689bf385f9 100644
--- a/cmd/ateapi/internal/store/storecontract/contract.go
+++ b/cmd/ateapi/internal/store/storecontract/contract.go
@@ -207,6 +207,7 @@ func RunContractTests(t *testing.T, setup func(t *testing.T) store.Interface) {
 	runAtespaceContractTests(t, setup)
 	runActorTemplateContractTests(t, setup)
 	runTagContractTests(t, setup)
+	runTagBorrowContractTests(t, setup)
 	runLeaseContractTests(t, setup)
 	runListOptionsContractTests(t, setup)
 	runUnknownFieldContractTests(t, setup)
@@ -2719,6 +2720,304 @@ func runAtespaceContractTests(t *testing.T, setup func(t *testing.T) store.Inter
 	})
 }
 
+func runTagBorrowContractTests(t *testing.T, setup func(t *testing.T) store.Interface) {
+	t.Helper()
+
+	const tagUID = "9a3c7e51-8b04-4f6d-a2e9-71c5d8f0b34a"
+	const otherTagUID = "2f8d14b6-5c93-4a70-be18-6d02a9e7c5f3"
+
+	borrowingActor := func(name, borrowedTagUID string) *ateapipb.Actor {
+		return &ateapipb.Actor{
+			Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name},
+			Status: &ateapipb.ActorStatus{
+				State:            ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
+				ExternalSnapshot: &ateapipb.ExternalSnapshot{SnapshotUri: testTagSnapshotURI("gs://bucket", testAtespace, borrowedTagUID)},
+			},
+		}
+	}
+
+	t.Run("TagBorrow_CreateActor", func(t *testing.T) {
+		tests := []struct {
+			name       string
+			snapshot   *ateapipb.ExternalSnapshot
+			wantBorrow bool
+		}{
+			{
+				name:       "a snapshot under the tag's prefix is a borrow",
+				snapshot:   &ateapipb.ExternalSnapshot{SnapshotUri: testTagSnapshotURI("gs://bucket", testAtespace, tagUID)},
+				wantBorrow: true,
+			},
+			{
+				name:       "a snapshot under the actor's own prefix is not",
+				snapshot:   &ateapipb.ExternalSnapshot{SnapshotUri: testActorSnapshotURI("gs://bucket", testAtespace, "snapshot-1")},
+				wantBorrow: false,
+			},
+			{
+				name:       "another tag's snapshot is not",
+				snapshot:   &ateapipb.ExternalSnapshot{SnapshotUri: testTagSnapshotURI("gs://bucket", testAtespace, otherTagUID)},
+				wantBorrow: false,
+			},
+			{
+				name:       "no snapshot at all is not",
+				snapshot:   nil,
+				wantBorrow: false,
+			},
+		}
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				s := setup(t)
+				mustCreateAtespace(t, s, testAtespace)
+
+				created, err := s.CreateActor(context.Background(), &ateapipb.Actor{
+					Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "session-1"},
+					Status: &ateapipb.ActorStatus{
+						State:            ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
+						ExternalSnapshot: test.snapshot,
+					},
+				})
+				if err != nil {
+					t.Fatalf("CreateActor failed: %v", err)
+				}
+				var want []string
+				if test.wantBorrow {
+					want = []string{created.GetMetadata().GetUid()}
+				}
+				page, err := s.ListTagBorrowers(context.Background(), tagUID, store.ListOptions{PageSize: 1000})
+				if err != nil {
+					t.Fatalf("ListTagBorrowers failed: %v", err)
+				}
+				if diff := cmp.Diff(want, page.Items); diff != "" {
+					t.Errorf("borrowers of the tag (-want +got):\n%s", diff)
+				}
+			})
+		}
+	})
+
+	t.Run("TagBorrow_UpdateActor", func(t *testing.T) {
+		tests := []struct {
+			name       string
+			mutate     func(*ateapipb.Actor)
+			wantBorrow bool
+		}{
+			{
+				name: "taking over the snapshot ends the borrow",
+				mutate: func(a *ateapipb.Actor) {
+					a.Status.ExternalSnapshot.SnapshotUri = testActorSnapshotURI("gs://bucket", testAtespace, "snapshot-1")
+				},
+				wantBorrow: false,
+			},
+			{
+				name:       "dropping the snapshot ends the borrow",
+				mutate:     func(a *ateapipb.Actor) { a.Status.ExternalSnapshot = nil },
+				wantBorrow: false,
+			},
+			{
+				name:       "a suspend that writes no snapshot leaves the borrow standing",
+				mutate:     func(a *ateapipb.Actor) { a.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED },
+				wantBorrow: true,
+			},
+			{
+				name: "moving to another tag moves the borrow",
+				mutate: func(a *ateapipb.Actor) {
+					a.Status.ExternalSnapshot.SnapshotUri = testTagSnapshotURI("gs://bucket", testAtespace, otherTagUID)
+				},
+				wantBorrow: false,
+			},
+		}
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				s := setup(t)
+				ctx := context.Background()
+				mustCreateAtespace(t, s, testAtespace)
+
+				created, err := s.CreateActor(ctx, borrowingActor("session-1", tagUID))
+				if err != nil {
+					t.Fatalf("CreateActor failed: %v", err)
+				}
+				actorRef := resources.ActorRefFromActor(created)
+				if _, err := s.UpdateActor(ctx, actorRef, store.PreconditionFrom(created), func(toUpdate *ateapipb.Actor) error {
+					test.mutate(toUpdate)
+					return nil
+				}); err != nil {
+					t.Fatalf("UpdateActor failed: %v", err)
+				}
+				var want []string
+				if test.wantBorrow {
+					want = []string{created.GetMetadata().GetUid()}
+				}
+				page, err := s.ListTagBorrowers(ctx, tagUID, store.ListOptions{PageSize: 1000})
+				if err != nil {
+					t.Fatalf("ListTagBorrowers failed: %v", err)
+				}
+				if diff := cmp.Diff(want, page.Items); diff != "" {
+					t.Errorf("borrowers of the tag (-want +got):\n%s", diff)
+				}
+			})
+		}
+	})
+
+	// A snapshot URI the store cannot read tells it nothing about whether a Tag
+	// lent the snapshot, so the write is refused rather than recorded as "no
+	// borrow". Storing the actor and dropping the borrow would leave DeleteTag
+	// free to destroy the snapshot the actor is still running on.
+	t.Run("TagBorrow_UnreadableSnapshotURI", func(t *testing.T) {
+		const unreadableURI = "not-a-valid-snapshot-uri"
+
+		t.Run("CreateActor is refused", func(t *testing.T) {
+			s := setup(t)
+			ctx := context.Background()
+			mustCreateAtespace(t, s, testAtespace)
+
+			actorRef := resources.ActorRef{Atespace: testAtespace, Name: "session-1"}
+			if _, err := s.CreateActor(ctx, &ateapipb.Actor{
+				Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name},
+				Status: &ateapipb.ActorStatus{
+					State:            ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
+					ExternalSnapshot: &ateapipb.ExternalSnapshot{SnapshotUri: unreadableURI},
+				},
+			}); err == nil {
+				t.Fatalf("CreateActor with snapshot URI %q = nil error, want the write refused", unreadableURI)
+			}
+			if _, err := s.GetActor(ctx, actorRef); !errors.Is(err, store.ErrNotFound) {
+				t.Errorf("GetActor after the refused create = %v, want ErrNotFound, the whole write to have rolled back", err)
+			}
+		})
+
+		t.Run("UpdateActor is refused and the borrow stands", func(t *testing.T) {
+			s := setup(t)
+			ctx := context.Background()
+			mustCreateAtespace(t, s, testAtespace)
+
+			created, err := s.CreateActor(ctx, borrowingActor("session-1", tagUID))
+			if err != nil {
+				t.Fatalf("CreateActor failed: %v", err)
+			}
+			actorRef := resources.ActorRefFromActor(created)
+			if _, err := s.UpdateActor(ctx, actorRef, store.PreconditionFrom(created), func(toUpdate *ateapipb.Actor) error {
+				toUpdate.Status.ExternalSnapshot.SnapshotUri = unreadableURI
+				return nil
+			}); err == nil {
+				t.Fatalf("UpdateActor to snapshot URI %q = nil error, want the write refused", unreadableURI)
+			}
+			page, err := s.ListTagBorrowers(ctx, tagUID, store.ListOptions{PageSize: 1000})
+			if err != nil {
+				t.Fatalf("ListTagBorrowers failed: %v", err)
+			}
+			want := []string{created.GetMetadata().GetUid()}
+			if diff := cmp.Diff(want, page.Items); diff != "" {
+				t.Errorf("borrowers after the refused update (-want +got), want the borrow to stand:\n%s", diff)
+			}
+		})
+	})
+
+	t.Run("TagBorrow_DeleteActor", func(t *testing.T) {
+		s := setup(t)
+		ctx := context.Background()
+		mustCreateAtespace(t, s, testAtespace)
+
+		created, err := s.CreateActor(ctx, borrowingActor("session-1", tagUID))
+		if err != nil {
+			t.Fatalf("CreateActor failed: %v", err)
+		}
+		actorRef := resources.ActorRefFromActor(created)
+		deleting, err := s.UpdateActor(ctx, actorRef, store.PreconditionFrom(created), func(toUpdate *ateapipb.Actor) error {
+			toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_DELETING
+			return nil
+		})
+		if err != nil {
+			t.Fatalf("marking the actor deleting failed: %v", err)
+		}
+		page, err := s.ListTagBorrowers(ctx, tagUID, store.ListOptions{PageSize: 1000})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if diff := cmp.Diff([]string{created.GetMetadata().GetUid()}, page.Items); diff != "" {
+			t.Fatalf("borrowers while the actor is deleting (-want +got), want the borrow to stand until the row is gone:\n%s", diff)
+		}
+		if _, err := s.DeleteActor(ctx, resources.ActorRefFromActor(deleting)); err != nil {
+			t.Fatalf("DeleteActor failed: %v", err)
+		}
+		page, err = s.ListTagBorrowers(ctx, tagUID, store.ListOptions{PageSize: 1000})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if page.Items != nil {
+			t.Errorf("borrowers after the actor was deleted = %v, want none", page.Items)
+		}
+	})
+
+	t.Run("TagBorrow_ListPagination", func(t *testing.T) {
+		s := setup(t)
+		ctx := context.Background()
+		mustCreateAtespace(t, s, testAtespace)
+
+		var created []*ateapipb.Actor
+		var wantBorrowers []string
+		for _, name := range []string{"session-1", "session-2", "session-3"} {
+			actor, err := s.CreateActor(ctx, borrowingActor(name, tagUID))
+			if err != nil {
+				t.Fatalf("CreateActor(%s) failed: %v", name, err)
+			}
+			created = append(created, actor)
+			wantBorrowers = append(wantBorrowers, actor.GetMetadata().GetUid())
+		}
+
+		// Walking one borrower at a time must yield every borrower exactly once,
+		// in UID order, and stop on its own.
+		var got []string
+		var pageToken string
+		for range len(created) {
+			page, err := s.ListTagBorrowers(ctx, tagUID, store.ListOptions{PageSize: 1, PageToken: pageToken})
+			if err != nil {
+				t.Fatalf("ListTagBorrowers failed: %v", err)
+			}
+			if len(page.Items) != 1 {
+				t.Fatalf("ListTagBorrowers(page size 1) returned %d borrowers, want 1", len(page.Items))
+			}
+			got = append(got, page.Items...)
+			pageToken = page.NextPageToken
+		}
+
+		slices.Sort(wantBorrowers)
+		if diff := cmp.Diff(wantBorrowers, got); diff != "" {
+			t.Errorf("borrowers walked one page at a time (-want +got):\n%s", diff)
+		}
+		if pageToken != "" {
+			t.Errorf("the page holding the last borrower carries NextPageToken %q, want the walk to end there", pageToken)
+		}
+	})
+
+	t.Run("TagBorrow_ListRejectsForeignPageToken", func(t *testing.T) {
+		s := setup(t)
+		ctx := context.Background()
+		mustCreateAtespace(t, s, testAtespace)
+
+		for _, name := range []string{"session-1", "session-2"} {
+			if _, err := s.CreateActor(ctx, borrowingActor(name, tagUID)); err != nil {
+				t.Fatalf("CreateActor(%s) failed: %v", name, err)
+			}
+		}
+		page, err := s.ListTagBorrowers(ctx, tagUID, store.ListOptions{PageSize: 1})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if _, err := s.ListTagBorrowers(ctx, otherTagUID, store.ListOptions{PageSize: 1, PageToken: page.NextPageToken}); !errors.Is(err, store.ErrInvalidPageToken) {
+			t.Errorf("ListTagBorrowers with another tag's page token = %v, want ErrInvalidPageToken", err)
+		}
+	})
+
+	t.Run("TagBorrow_UnknownTag", func(t *testing.T) {
+		s := setup(t)
+		page, err := s.ListTagBorrowers(context.Background(), tagUID, store.ListOptions{PageSize: 10})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if len(page.Items) != 0 || page.HasNextPage() {
+			t.Errorf("ListTagBorrowers on a tag nobody borrows = %v (next token %q), want empty and no next page", page.Items, page.NextPageToken)
+		}
+	})
+}
+
 func runLeaseContractTests(t *testing.T, setup func(t *testing.T) store.Interface) {
 	t.Helper()
 
diff --git a/internal/resources/snapshot_test.go b/internal/resources/snapshot_test.go
index 195ef8da90..1b7de8f466 100644
--- a/internal/resources/snapshot_test.go
+++ b/internal/resources/snapshot_test.go
@@ -237,6 +237,41 @@ func TestNewTagSnapshotURI(t *testing.T) {
 	}
 }
 
+// TestSnapshotOwnerTagUID covers reading a lender out of a parsed URI. An
+// Actor records the URI it was cloned from, and its owner is all the deletion
+// check has to tell a borrowed tag snapshot from one the Actor took itself.
+func TestSnapshotOwnerTagUID(t *testing.T) {
+	tests := []struct {
+		name    string
+		owner   SnapshotOwner
+		want    string
+		wantTag bool
+	}{
+		{
+			name:    "a tag owns its snapshot",
+			owner:   TagSnapshotOwner("team-a", "tag-uid-1"),
+			want:    "tag-uid-1",
+			wantTag: true,
+		},
+		{
+			name:  "an actor's own snapshot has no tag",
+			owner: ActorSnapshotOwner("team-a", "actor-uid"),
+		},
+		{
+			name:  "the zero owner has no tag",
+			owner: SnapshotOwner{},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, gotTag := tc.owner.TagUID()
+			if got != tc.want || gotTag != tc.wantTag {
+				t.Errorf("%s.TagUID() = (%q, %t), want (%q, %t)", tc.owner, got, gotTag, tc.want, tc.wantTag)
+			}
+		})
+	}
+}
+
 // TestSnapshotOwnedBy covers the check every collector makes before deleting.
 // An owner may only reach its own objects, which is what keeps an actor
 // borrowing a tag's snapshot from collecting it out from under every other

From d8e1c6208e9d9c0f7f5f01a96a766e0f6c477a65 Mon Sep 17 00:00:00 2001
From: Luiz Oliveira <luizaoj@google.com>
Date: Wed, 16 Sep 2026 16:08:46 -0400
Subject: [PATCH 2/6] Refuse to delete a Tag an Actor is still borrowing

---
 .../controlapi/functionaltest/actor_test.go   |  10 ++
 .../controlapi/functionaltest/tag_test.go     | 154 ++++++++++++++++++
 cmd/ateapi/internal/controlapi/service.go     |   1 +
 cmd/ateapi/internal/controlapi/tag.go         |  25 ++-
 cmd/ateapi/internal/controlapi/tag_test.go    |  62 +++++++
 5 files changed, 249 insertions(+), 3 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go
index 07ce4cd75a..847973a945 100644
--- a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go
+++ b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go
@@ -2357,6 +2357,16 @@ func TestSuspendActor(t *testing.T) {
 	assertSnapshotCollected(t, tc, snapshotURI)
 	assertSnapshotPresent(t, tc, tagSnapshotURI)
 
+	// The cross-atespace clone never suspended, so the tag's snapshot is still
+	// its starting state. A borrow holds the tag back from whichever atespace it
+	// was taken out in.
+	if _, err := tc.client.DeleteTag(context.Background(), &ateapipb.DeleteTagRequest{Tag: tagRef}); status.Code(err) != codes.FailedPrecondition {
+		t.Fatalf("DeleteTag while other/cross-atespace borrows it = %v, want FailedPrecondition", err)
+	}
+	if _, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{Actor: &ateapipb.ObjectRef{Atespace: "other", Name: "cross-atespace"}}); err != nil {
+		t.Fatalf("DeleteActor(other/cross-atespace) failed: %v", err)
+	}
+
 	if deleted, err := tc.client.DeleteTag(context.Background(), &ateapipb.DeleteTagRequest{Tag: tagRef}); err != nil || deleted.GetMetadata().GetName() != tagRef.GetName() {
 		t.Fatalf("DeleteTag = (%v, %v)", deleted, err)
 	}
diff --git a/cmd/ateapi/internal/controlapi/functionaltest/tag_test.go b/cmd/ateapi/internal/controlapi/functionaltest/tag_test.go
index 8f9be6ed3e..ae7e8aa1fc 100644
--- a/cmd/ateapi/internal/controlapi/functionaltest/tag_test.go
+++ b/cmd/ateapi/internal/controlapi/functionaltest/tag_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/testing/protocmp"
 )
 
@@ -175,6 +176,14 @@ func suspendActorForTest(t *testing.T, tc *testContext, workerName, name string)
 	}}); err != nil {
 		t.Fatalf("CreateActor(%s) failed: %v", name, err)
 	}
+	return runAndSuspendActorForTest(t, tc, workerName, name)
+}
+
+// runAndSuspendActorForTest resumes an existing actor on workerName and
+// suspends it, returning the URI of the external snapshot the suspend wrote.
+func runAndSuspendActorForTest(t *testing.T, tc *testContext, workerName, name string) string {
+	t.Helper()
+	ctx := context.Background()
 	// Successive actors share the one worker, and scheduling reads the worker
 	// cache: the preceding suspend released the worker in the store, but the
 	// cache only learns of it on its next watch poll.
@@ -197,6 +206,151 @@ func suspendActorForTest(t *testing.T, tc *testContext, workerName, name string)
 	return uri
 }
 
+// TestDeleteTag_RefusedWhileCloneBorrowsSnapshot walks the whole loop over the
+// wire: a tag cannot be deleted while an actor cloned from it is still running
+// on the tag's snapshot, and can be once that clone has suspended into one of
+// its own and resumed off that one.
+func TestDeleteTag_RefusedWhileCloneBorrowsSnapshot(t *testing.T) {
+	ns := namespaceForTest("ns-delete-tag-borrowed")
+	tc := setupTest(t, ns)
+	defer tc.cleanup()
+
+	ctx := context.Background()
+	createTemplate(t, tc, ns)
+	workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
+	suspendActorForTest(t, tc, workerName, "actor-a")
+
+	const tagName = "v1"
+	tagRef := &ateapipb.ObjectRef{Atespace: testAtespace, Name: tagName}
+	tag, err := tc.client.CreateTag(ctx, &ateapipb.CreateTagRequest{
+		Tag: &ateapipb.Tag{
+			Metadata:    &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: tagName},
+			Scope:       ateapipb.TagScope_TAG_SCOPE_ATESPACE,
+			SourceActor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-a"},
+		},
+	})
+	if err != nil {
+		t.Fatalf("CreateTag failed: %v", err)
+	}
+	tagSnapshotURI := tag.GetStatus().GetSnapshot().GetSnapshotUri()
+
+	if _, err := tc.client.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{
+		Metadata:      &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "clone-1"},
+		ActorTemplate: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "tmpl1"},
+		SourceTag:     tagRef,
+	}}); err != nil {
+		t.Fatalf("CreateActor(clone-1) from tag %s failed: %v", tagName, err)
+	}
+
+	_, err = tc.client.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef})
+	if got, want := status.Code(err), codes.FailedPrecondition; got != want {
+		t.Fatalf("DeleteTag while clone-1 borrows its snapshot = %v (code %v), want %v", err, got, want)
+	}
+	if _, err := tc.client.GetTag(ctx, &ateapipb.GetTagRequest{Tag: tagRef}); err != nil {
+		t.Fatalf("GetTag after the refusal: %v", err)
+	}
+	if got := snapshotObjectNames(t, tc, tagSnapshotURI); len(got) == 0 {
+		t.Error("the refused delete collected the tag's external snapshot anyway")
+	}
+
+	// The clone's first suspend writes a snapshot of its own, which is what
+	// ends the borrow and frees the tag.
+	cloneSnapshotURI := runAndSuspendActorForTest(t, tc, workerName, "clone-1")
+	if cloneSnapshotURI == tagSnapshotURI {
+		t.Fatalf("clone-1 suspended back into the tag's snapshot %s", cloneSnapshotURI)
+	}
+
+	// Resume actor: check that the tag can be deleted, while the actor is up,
+	// since the actor no longer borrows the tag's snapshot.
+	waitForWorkerAvailable(t, tc, workerName)
+	resumed, err := tc.client.ResumeActor(ctx, &ateapipb.ResumeActorRequest{Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "clone-1"}})
+	if err != nil {
+		t.Fatalf("ResumeActor(clone-1) failed: %v", err)
+	}
+	if got := resumed.GetActor().GetStatus().GetExternalSnapshot().GetSnapshotUri(); got != cloneSnapshotURI {
+		t.Fatalf("clone-1 resumed holding snapshot %q, want its own %q", got, cloneSnapshotURI)
+	}
+	if got, want := resumed.GetActor().GetStatus().GetState(), ateapipb.ActorState_ACTOR_STATE_RUNNING; got != want {
+		t.Fatalf("clone-1 state after the resume = %v, want %v", got, want)
+	}
+
+	if _, err := tc.client.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef}); err != nil {
+		t.Fatalf("DeleteTag once the borrow ended: %v", err)
+	}
+	if _, err := tc.client.GetTag(ctx, &ateapipb.GetTagRequest{Tag: tagRef}); status.Code(err) != codes.NotFound {
+		t.Errorf("GetTag after the delete = %v, want NotFound", err)
+	}
+	if got := snapshotObjectNames(t, tc, cloneSnapshotURI); len(got) == 0 {
+		t.Error("deleting the tag collected the clone's own external snapshot")
+	}
+}
+
+func TestDeleteTag_RefusedWhileGoldenCloneBorrowsSnapshot(t *testing.T) {
+	ns := namespaceForTest("ns-delete-golden-tag-borrowed")
+	tc := setupTest(t, ns)
+	defer tc.cleanup()
+
+	ctx := context.Background()
+	template := createTemplate(t, tc, ns)
+	workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
+	templateRef := resources.ActorTemplateRefFromActorTemplate(template).ToObjectRef()
+	goldenRef := template.GetStatus().GetGoldenSnapshotStatus().GetGoldenTag()
+
+	created, err := tc.client.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{
+		Metadata:      &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "actor-a"},
+		ActorTemplate: templateRef,
+	}})
+	if err != nil {
+		t.Fatalf("CreateActor(actor-a) failed: %v", err)
+	}
+	goldenURI := goldenSnapshotURI(t, tc, template)
+	if got := created.GetStatus().GetExternalSnapshot().GetSnapshotUri(); got != goldenURI {
+		t.Fatalf("actor-a was created holding snapshot %q, want the template's golden %q", got, goldenURI)
+	}
+
+	_, err = tc.client.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: goldenRef})
+	if got, want := status.Code(err), codes.FailedPrecondition; got != want {
+		t.Fatalf("DeleteTag(golden) while actor-a borrows its snapshot = %v (code %v), want %v", err, got, want)
+	}
+	// Deleting the template is the way a golden tag is normally collected, so
+	// the borrow has to hold that back too, leaving both resources in place.
+	_, err = tc.client.DeleteActorTemplate(ctx, &ateapipb.DeleteActorTemplateRequest{ActorTemplate: templateRef})
+	if got, want := status.Code(err), codes.FailedPrecondition; got != want {
+		t.Fatalf("DeleteActorTemplate while actor-a borrows the golden snapshot = %v (code %v), want %v", err, got, want)
+	}
+	if _, err := tc.client.GetActorTemplate(ctx, &ateapipb.GetActorTemplateRequest{ActorTemplate: templateRef}); err != nil {
+		t.Fatalf("GetActorTemplate after the refusal: %v", err)
+	}
+	if _, err := tc.client.GetTag(ctx, &ateapipb.GetTagRequest{Tag: goldenRef}); err != nil {
+		t.Fatalf("GetTag(golden) after the refusal: %v", err)
+	}
+	assertSnapshotPresent(t, tc, goldenURI)
+
+	// The first suspend writes a snapshot under the actor's own prefix and the
+	// resume restores that one, so the golden tag is free from the suspend on.
+	ownSnapshotURI := runAndSuspendActorForTest(t, tc, workerName, "actor-a")
+	assertSnapshotOwnedByActor(t, created, ownSnapshotURI)
+	waitForWorkerAvailable(t, tc, workerName)
+	resumed, err := tc.client.ResumeActor(ctx, &ateapipb.ResumeActorRequest{Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-a"}})
+	if err != nil {
+		t.Fatalf("ResumeActor(actor-a) failed: %v", err)
+	}
+	if got := resumed.GetActor().GetStatus().GetExternalSnapshot().GetSnapshotUri(); got != ownSnapshotURI {
+		t.Fatalf("actor-a resumed holding snapshot %q, want its own %q", got, ownSnapshotURI)
+	}
+
+	if _, err := tc.client.DeleteActorTemplate(ctx, &ateapipb.DeleteActorTemplateRequest{ActorTemplate: templateRef}); err != nil {
+		t.Fatalf("DeleteActorTemplate once the borrow ended: %v", err)
+	}
+	if _, err := tc.client.GetTag(ctx, &ateapipb.GetTagRequest{Tag: goldenRef}); status.Code(err) != codes.NotFound {
+		t.Errorf("GetTag(golden) after the delete = %v, want NotFound", err)
+	}
+	assertSnapshotCollected(t, tc, goldenURI)
+	if got := snapshotObjectNames(t, tc, ownSnapshotURI); len(got) == 0 {
+		t.Error("deleting the template collected the actor's own external snapshot")
+	}
+}
+
 // TestUpdateTag_Preconditions verifies the required version and uid
 // guards carried in the tag's metadata.
 func TestUpdateTag_Preconditions(t *testing.T) {
diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go
index b433515586..2b81af7103 100644
--- a/cmd/ateapi/internal/controlapi/service.go
+++ b/cmd/ateapi/internal/controlapi/service.go
@@ -121,6 +121,7 @@ type serviceStore interface {
 	ListTags(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Tag], error)
 	UpdateTag(ctx context.Context, tagRef resources.TagRef, precondition store.Precondition, mutate func(toUpdate *ateapipb.Tag) error) (*ateapipb.Tag, error)
 	DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error)
+	ListTagBorrowers(ctx context.Context, tagUID string, opts store.ListOptions) (store.ListResponse[string], error)
 	CreateAtespace(ctx context.Context, atespace *ateapipb.Atespace) (*ateapipb.Atespace, error)
 	GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error)
 	ListAtespaces(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Atespace], error)
diff --git a/cmd/ateapi/internal/controlapi/tag.go b/cmd/ateapi/internal/controlapi/tag.go
index a5ced8e6dd..46ed63ac3c 100644
--- a/cmd/ateapi/internal/controlapi/tag.go
+++ b/cmd/ateapi/internal/controlapi/tag.go
@@ -255,9 +255,8 @@ func ValidateCustom_UpdateTagRequest_Tag(ctx context.Context, op operation.Opera
 // CreateActor racing this delete can seed an Actor from content that is going
 // away. That race is accepted for now.
 //
-// Note that this destroys the external snapshot: an Actor created from the tag
-// and never suspended is still borrowing it and becomes unrecoverable. Do not
-// delete a tag while clones of it exist.
+// DeletaTag is refused with FailedPrecondition if at least one actor is still borrowing
+// the tag's snapshot.
 func (s *RPCService) DeleteTag(ctx context.Context, req *ateapipb.DeleteTagRequest) (*ateapipb.Tag, error) {
 	// TODO: mode delete orchestration to a workflow.
 	if errs := validateDeleteTagRequest(ctx, req); len(errs) > 0 {
@@ -280,6 +279,9 @@ func (s *RPCService) DeleteTag(ctx context.Context, req *ateapipb.DeleteTagReque
 		}
 		return nil, fmt.Errorf("while getting tag: %w", err)
 	}
+	if err := s.checkTagBorrowers(ctx, stored); err != nil {
+		return nil, err
+	}
 	if err := s.releaseTagSnapshot(ctx, stored); err != nil {
 		return nil, err
 	}
@@ -294,6 +296,23 @@ func (s *RPCService) DeleteTag(ctx context.Context, req *ateapipb.DeleteTagReque
 	return tag, nil
 }
 
+// checkTagBorrowers refuses the delete while an Actor is still using the tag's
+// external snapshot as its own.
+func (s *RPCService) checkTagBorrowers(ctx context.Context, tag *ateapipb.Tag) error {
+	atespace, name := tag.GetMetadata().GetAtespace(), tag.GetMetadata().GetName()
+
+	borrowers, err := s.impl.ListTagBorrowers(ctx, tag.GetMetadata().GetUid(), store.ListOptions{PageSize: 1})
+	if err != nil {
+		return fmt.Errorf("while listing the borrowers of tag %s/%s: %w", atespace, name, err)
+	}
+	if len(borrowers.Items) == 0 {
+		return nil
+	}
+	return status.Errorf(codes.FailedPrecondition,
+		"Tag %s/%s cannot be deleted because its snapshot is still in use by at least one Actor created from it",
+		atespace, name)
+}
+
 // releaseTagSnapshot deletes the objects the tag's external snapshot is made
 // of. It tolerates a partly-collected snapshot, so a retry finishes cleanly.
 // It collects the in-progress snapshot too.
diff --git a/cmd/ateapi/internal/controlapi/tag_test.go b/cmd/ateapi/internal/controlapi/tag_test.go
index 69c6ad52d3..8d47e04ee4 100644
--- a/cmd/ateapi/internal/controlapi/tag_test.go
+++ b/cmd/ateapi/internal/controlapi/tag_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/testing/protocmp"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 
@@ -981,6 +982,67 @@ func TestDeleteTag_ReleasesExternalSnapshot(t *testing.T) {
 	}
 }
 
+// seedTagBorrower stores a suspended actor running on the tag's external
+// snapshot rather than one of its own, which is what CreateActor leaves behind
+// for an actor cloned from a tag.
+func seedTagBorrower(t *testing.T, ctx context.Context, persistence store.Interface, tag *ateapipb.Tag, name string) *ateapipb.Actor {
+	t.Helper()
+	atespace := tag.GetMetadata().GetAtespace()
+	return storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{
+		Metadata:      &ateapipb.ResourceMetadata{Atespace: atespace, Name: name},
+		ActorTemplate: &ateapipb.ObjectRef{Atespace: atespace, Name: "sub-tmpl"},
+		SourceTag:     resources.TagRefFromTag(tag).ToObjectRef(),
+		Status: &ateapipb.ActorStatus{
+			State:            ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
+			ExternalSnapshot: proto.CloneOf(tag.GetStatus().GetSnapshot()),
+		},
+	})
+}
+
+// TestDeleteTag_RefusesWhileBorrowed verifies the delete is refused while an
+// actor is still running on the tag's snapshot, leaving both the row and the
+// objects alone, and goes through once that actor holds a snapshot of its own.
+func TestDeleteTag_RefusesWhileBorrowed(t *testing.T) {
+	ctx := context.Background()
+	persistence := newTestPersistence(t)
+	template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl")
+	w, objects := newFinalizeWorkflow(persistence)
+	source, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst")
+	tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(source), "v1"))
+	if err != nil {
+		t.Fatalf("TagActorSnapshot: %v", err)
+	}
+	tagRef := resources.TagRefFromTag(tag)
+	uri := mustReservedTagSnapshotURI(t, tag)
+	borrower := seedTagBorrower(t, ctx, persistence, tag, "clone-1")
+	svc := &RPCService{impl: newServiceImpl(persistence, nil), objectStore: objects}
+	req := &ateapipb.DeleteTagRequest{Tag: tagRef.ToObjectRef()}
+
+	_, err = svc.DeleteTag(ctx, req)
+	if got, want := status.Code(err), codes.FailedPrecondition; got != want {
+		t.Fatalf("DeleteTag = %v (code %v), want %v", err, got, want)
+	}
+	if _, err := persistence.GetTag(ctx, tagRef); err != nil {
+		t.Fatalf("GetTag after the refusal: %v", err)
+	}
+	if got := objects.Snapshot(t, uri); len(got) == 0 {
+		t.Error("the refused delete collected the tag's external snapshot anyway")
+	}
+
+	// The borrower suspends and writes a snapshot under its own prefix, which is
+	// what ends the borrow.
+	own := mustActorSnapshotURI(t, template, borrower, "clone-1-snapshot")
+	mustUpdateActorStatus(t, ctx, persistence, borrower, func(s *ateapipb.ActorStatus) {
+		s.ExternalSnapshot = &ateapipb.ExternalSnapshot{SnapshotUri: own.String(), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL}
+	})
+	if _, err := svc.DeleteTag(ctx, req); err != nil {
+		t.Fatalf("DeleteTag once the borrow ended: %v", err)
+	}
+	if _, err := persistence.GetTag(ctx, tagRef); !errors.Is(err, store.ErrNotFound) {
+		t.Errorf("GetTag after the delete = %v, want ErrNotFound", err)
+	}
+}
+
 // TestDeleteTag_ReleasesPendingSnapshot verifies that deleting a
 // tag whose create never finished collects what that create stranded. The
 // pending row names the prefix the copy was writing into, and it is the only

From 4a8a3c6f63062a19d563e2af4ebf9e9b5c1ec9a5 Mon Sep 17 00:00:00 2001
From: Luiz Oliveira <luizaoj@google.com>
Date: Thu, 17 Sep 2026 16:31:20 -0400
Subject: [PATCH 3/6] Update docs

---
 benchmarking/locust/common/ateapi_pb2_grpc.py |  6 +++---
 docs/api-guide.md                             |  2 +-
 docs/architecture.md                          |  3 ++-
 pkg/proto/ateapipb/ateapi.proto               |  6 +++---
 pkg/proto/ateapipb/ateapi_grpc.pb.go          | 12 ++++++------
 5 files changed, 15 insertions(+), 14 deletions(-)

diff --git a/benchmarking/locust/common/ateapi_pb2_grpc.py b/benchmarking/locust/common/ateapi_pb2_grpc.py
index 97ef55f6fb..b777a6a87a 100644
--- a/benchmarking/locust/common/ateapi_pb2_grpc.py
+++ b/benchmarking/locust/common/ateapi_pb2_grpc.py
@@ -357,9 +357,9 @@ def UpdateTag(self, request, context):
         raise NotImplementedError('Method not implemented!')
 
     def DeleteTag(self, request, context):
-        """Delete a Tag and the external snapshot it owns. Actors created from the
-        tag that have not yet been suspended still point at that external snapshot
-        and become unrecoverable, so do not delete a tag while such Actors exist.
+        """Delete a Tag and the external snapshot it owns. Rejects
+        (FailedPrecondition) while an Actor created from the tag is still
+        borrowing that snapshot, which it does until its own first suspend.
         """
         context.set_code(grpc.StatusCode.UNIMPLEMENTED)
         context.set_details('Method not implemented!')
diff --git a/docs/api-guide.md b/docs/api-guide.md
index b6d185006b..e9e26c41fa 100644
--- a/docs/api-guide.md
+++ b/docs/api-guide.md
@@ -502,7 +502,7 @@ An actor created from a tag borrows the tag's copy instead of taking one of its
 
 Deletion always runs before the database reference is dropped, and a failure fails the whole RPC. Clients are expected to retry with the same arguments: destinations are deterministic and every phase tolerates a partly-completed predecessor, so a retry resumes rather than duplicating work. The cost of that ordering is that a crash between the two can leave an external snapshot no row names; the reverse order would instead lose the handle needed to ever delete it.
 
-> **Do not delete a tag while actors created from it exist.** A clone borrows the tag's snapshot rather than copying it, and only stops borrowing at its own first suspend (its `status.externalSnapshot.snapshotUri` still names the tag's prefix while it is). Deleting the tag leaves such a clone unable to resume. This is not prevented today.
+> **A tag cannot be deleted while an actor is still borrowing its snapshot.** A clone borrows the tag's snapshot rather than copying it, and only stops borrowing at its own first suspend (its `status.externalSnapshot.snapshotUri` still names the tag's prefix while it is). Deleting the tag then would leave the clone unable to resume, so `DeleteTag` rejects with `FAILED_PRECONDITION` while any borrower remains. Suspend each borrowing actor, so that it takes a snapshot of its own, or delete it, and the tag becomes deletable.
 
 #### `DeleteActor`
 Removes an actor from the registry and cleans up associated resources.
diff --git a/docs/architecture.md b/docs/architecture.md
index e0b06a1343..72058548c4 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -468,7 +468,8 @@ Snapshots may be given tags owned and addressed by an Atespace. The same tag
 name may exist in different Atespaces. A tag is an immutable alias and retention pin:
 it holds its own copy of the external snapshot, made at creation, so it outlives the 
 Actor that took it, and publishing it permits reuse from other Atespaces without changing its
-`atespace/name` address. Deleting a tag deletes that copy; an Atespace with
+`atespace/name` address. Deleting a tag deletes that copy, so it is refused
+while an Actor created from the tag is still borrowing it; an Atespace with
 tags cannot be deleted until they are.
 
 ### Phase 4: Deletion
diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto
index 0022537716..355ecae877 100644
--- a/pkg/proto/ateapipb/ateapi.proto
+++ b/pkg/proto/ateapipb/ateapi.proto
@@ -86,9 +86,9 @@ service Control {
   // Publish or unpublish a Tag without changing its address.
   rpc UpdateTag(UpdateTagRequest) returns (Tag) {}
 
-  // Delete a Tag and the external snapshot it owns. Actors created from the
-  // tag that have not yet been suspended still point at that external snapshot
-  // and become unrecoverable, so do not delete a tag while such Actors exist.
+  // Delete a Tag and the external snapshot it owns. Rejects
+  // (FailedPrecondition) while an Actor created from the tag is still
+  // borrowing that snapshot, which it does until its own first suspend.
   rpc DeleteTag(DeleteTagRequest) returns (Tag) {}
 
   // List Workers.
diff --git a/pkg/proto/ateapipb/ateapi_grpc.pb.go b/pkg/proto/ateapipb/ateapi_grpc.pb.go
index f88dbd6700..8b99dc144d 100644
--- a/pkg/proto/ateapipb/ateapi_grpc.pb.go
+++ b/pkg/proto/ateapipb/ateapi_grpc.pb.go
@@ -121,9 +121,9 @@ type ControlClient interface {
 	ListTags(ctx context.Context, in *ListTagsRequest, opts ...grpc.CallOption) (*ListTagsResponse, error)
 	// Publish or unpublish a Tag without changing its address.
 	UpdateTag(ctx context.Context, in *UpdateTagRequest, opts ...grpc.CallOption) (*Tag, error)
-	// Delete a Tag and the external snapshot it owns. Actors created from the
-	// tag that have not yet been suspended still point at that external snapshot
-	// and become unrecoverable, so do not delete a tag while such Actors exist.
+	// Delete a Tag and the external snapshot it owns. Rejects
+	// (FailedPrecondition) while an Actor created from the tag is still
+	// borrowing that snapshot, which it does until its own first suspend.
 	DeleteTag(ctx context.Context, in *DeleteTagRequest, opts ...grpc.CallOption) (*Tag, error)
 	// List Workers.
 	ListWorkers(ctx context.Context, in *ListWorkersRequest, opts ...grpc.CallOption) (*ListWorkersResponse, error)
@@ -561,9 +561,9 @@ type ControlServer interface {
 	ListTags(context.Context, *ListTagsRequest) (*ListTagsResponse, error)
 	// Publish or unpublish a Tag without changing its address.
 	UpdateTag(context.Context, *UpdateTagRequest) (*Tag, error)
-	// Delete a Tag and the external snapshot it owns. Actors created from the
-	// tag that have not yet been suspended still point at that external snapshot
-	// and become unrecoverable, so do not delete a tag while such Actors exist.
+	// Delete a Tag and the external snapshot it owns. Rejects
+	// (FailedPrecondition) while an Actor created from the tag is still
+	// borrowing that snapshot, which it does until its own first suspend.
 	DeleteTag(context.Context, *DeleteTagRequest) (*Tag, error)
 	// List Workers.
 	ListWorkers(context.Context, *ListWorkersRequest) (*ListWorkersResponse, error)

From cb6ac2534785d9df0fc71b421c703d1e49f6c968 Mon Sep 17 00:00:00 2001
From: Luiz Oliveira <luizaoj@google.com>
Date: Thu, 17 Sep 2026 16:50:20 -0400
Subject: [PATCH 4/6] Suspend actors before deleting the tag in e2e tests

---
 internal/e2e/suites/demo/demo_test.go | 23 +++++++++++++++++++----
 1 file changed, 19 insertions(+), 4 deletions(-)

diff --git a/internal/e2e/suites/demo/demo_test.go b/internal/e2e/suites/demo/demo_test.go
index 392128152f..0ffdc74e59 100644
--- a/internal/e2e/suites/demo/demo_test.go
+++ b/internal/e2e/suites/demo/demo_test.go
@@ -99,6 +99,12 @@ func TestActorSnapshotLifecycle(t *testing.T) {
 
 	sourceName := "snapshot-source-" + nsObj.Name
 	cloneName := "snapshot-clone-" + nsObj.Name
+	tagRef := &ateapipb.ObjectRef{Atespace: demoAtespace, Name: "e2e-" + nsObj.Name}
+	// Registered first so it runs last: the Tag cannot be deleted while an Actor
+	// is still borrowing its snapshot.
+	t.Cleanup(func() {
+		_, _ = clients.SubstrateAPI.DeleteTag(context.Background(), &ateapipb.DeleteTagRequest{Tag: tagRef})
+	})
 	for _, name := range []string{sourceName, cloneName} {
 		t.Cleanup(func() {
 			cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
@@ -124,10 +130,6 @@ func TestActorSnapshotLifecycle(t *testing.T) {
 	}
 	validateCounterResponse(t, response, "source", 1, 1)
 
-	tagRef := &ateapipb.ObjectRef{Atespace: demoAtespace, Name: "e2e-" + nsObj.Name}
-	t.Cleanup(func() {
-		_, _ = clients.SubstrateAPI.DeleteTag(context.Background(), &ateapipb.DeleteTagRequest{Tag: tagRef})
-	})
 	suspended, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
 		Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: sourceName},
 	})
@@ -201,6 +203,19 @@ func TestActorSnapshotLifecycle(t *testing.T) {
 	}
 	validateCounterResponse(t, response, "clone", 2, 2)
 
+	// The clone is running off the Tag's snapshot, so the Tag cannot be deleted
+	// out from under it.
+	if _, err := clients.SubstrateAPI.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef}); status.Code(err) != codes.FailedPrecondition {
+		t.Fatalf("DeleteTag while the clone borrows the snapshot returned %v, want FailedPrecondition", err)
+	}
+
+	// Suspending the clone writes it a snapshot of its own, which releases the
+	// Tag's.
+	if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
+		Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: cloneName},
+	}); err != nil {
+		t.Fatalf("failed to suspend cloned Actor: %v", err)
+	}
 	if _, err := clients.SubstrateAPI.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef}); err != nil {
 		t.Fatalf("failed to delete Tag: %v", err)
 	}

From 20d2548f26a5054088265e1f5936ac0b8f173c16 Mon Sep 17 00:00:00 2001
From: Luiz Oliveira <luizaoj@google.com>
Date: Fri, 18 Sep 2026 16:25:06 -0400
Subject: [PATCH 5/6] Fail closed if we fail to parse snapshot URI

---
 .../controlapi/functionaltest/actor_test.go   | 10 +++---
 .../controlapi/functionaltest/common_test.go  | 32 +++++++++++++++----
 .../controlapi/workflow_resume_test.go        | 11 -------
 internal/resources/snapshot.go                | 10 ++++++
 4 files changed, 40 insertions(+), 23 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go
index 847973a945..4c655e21a5 100644
--- a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go
+++ b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go
@@ -72,7 +72,7 @@ func TestCreateActor_Success(t *testing.T) {
 		Status: &ateapipb.ActorStatus{
 			State:                   ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
 			CurrentActorTemplateUid: tmpl.GetMetadata().GetUid(),
-			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
+			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t, tc, tmpl), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
 		},
 		WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "free"}},
 	}
@@ -694,7 +694,7 @@ func TestUpdateActor_Success(t *testing.T) {
 		Status: &ateapipb.ActorStatus{
 			State:                   ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
 			CurrentActorTemplateUid: tmpl.GetMetadata().GetUid(),
-			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
+			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t, tc, tmpl), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
 		},
 		WorkerSelector: &ateapipb.Selector{
 			MatchLabels: map[string]string{"tier": "paid"},
@@ -841,7 +841,7 @@ func TestUpdateActor(t *testing.T) {
 		Status: &ateapipb.ActorStatus{
 			State:                   ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
 			CurrentActorTemplateUid: tmpl.GetMetadata().GetUid(),
-			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
+			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t, tc, tmpl), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
 		},
 		WorkerSelector: &ateapipb.Selector{
 			MatchLabels: map[string]string{"tier": "paid"},
@@ -1812,7 +1812,7 @@ func TestResumeActor(t *testing.T) {
 		Status: &ateapipb.ActorStatus{
 			State:                   ateapipb.ActorState_ACTOR_STATE_RUNNING,
 			CurrentActorTemplateUid: tmpl.GetMetadata().GetUid(),
-			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
+			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t, tc, tmpl), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
 			WorkerAssignment: &ateapipb.WorkerAssignment{
 				Worker:          &ateapipb.ObjectRef{Name: podUID},
 				WorkerNamespace: ns,
@@ -2567,7 +2567,7 @@ func TestPauseActor(t *testing.T) {
 				ContentScope:              ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
 			},
 			CurrentActorTemplateUid: tmpl.GetMetadata().GetUid(),
-			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
+			ExternalSnapshot:        &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t, tc, tmpl), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
 		},
 	}
 
diff --git a/cmd/ateapi/internal/controlapi/functionaltest/common_test.go b/cmd/ateapi/internal/controlapi/functionaltest/common_test.go
index 4a62d08a51..f99d24762c 100644
--- a/cmd/ateapi/internal/controlapi/functionaltest/common_test.go
+++ b/cmd/ateapi/internal/controlapi/functionaltest/common_test.go
@@ -341,15 +341,16 @@ func assertSnapshotCollected(t *testing.T, tc *testContext, snapshotURI string)
 	}
 }
 
-// goldenSnapshotURI is the snapshot owned by the test template's golden tag.
-func goldenSnapshotURI(t *testing.T) string {
+// goldenSnapshotURI is the snapshot owned by the golden tag of tmpl. An Actor
+// created from tmpl with no source tag of its own starts out on it.
+func goldenSnapshotURI(t *testing.T, tc *testContext, tmpl *ateapipb.ActorTemplate) string {
 	t.Helper()
-	const goldenSnapshotName = "9c2f7b41-6d05-4e83-a1f7-3b8c0d5e2a94"
-	uri, err := resources.NewTagSnapshotURI(testStorageLocation, resources.GoldenActorAtespace, goldenSnapshotName)
+	goldenRef := tmpl.GetStatus().GetGoldenSnapshotStatus().GetGoldenTag()
+	tag, err := tc.client.GetTag(context.Background(), &ateapipb.GetTagRequest{Tag: goldenRef})
 	if err != nil {
-		t.Fatalf("NewTagSnapshotURI: %v", err)
+		t.Fatalf("GetTag(golden tag of %s/%s): %v", tmpl.GetMetadata().GetAtespace(), tmpl.GetMetadata().GetName(), err)
 	}
-	return uri.String()
+	return tag.GetStatus().GetSnapshot().GetSnapshotUri()
 }
 
 // snapshotOwnedByActor reports whether snapshotURI sits under the actor's own
@@ -459,14 +460,31 @@ func createTemplateWithContainersAndVolumes(t *testing.T, tc *testContext, ns st
 		SourceActor: &ateapipb.ObjectRef{Atespace: resources.GoldenActorAtespace, Name: created.GetMetadata().GetUid()},
 		Scope:       ateapipb.TagScope_TAG_SCOPE_PUBLISHED,
 		Status: &ateapipb.TagStatus{
-			Snapshot:         &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL},
 			ActorTemplateUid: created.GetMetadata().GetUid(),
 			SourceActorUid:   "9c2f7b41-6d05-4e83-a1f7-3b8c0d5e2a94",
+			StorageLocation:  testStorageLocation,
 		},
 	})
 	if err != nil {
 		t.Fatalf("create golden tag: %v", err)
 	}
+	// The golden snapshot sits under the tag's own prefix, keyed on the UID the
+	// store assigns, so it can only be recorded once the row exists. That is
+	// what makes an Actor that inherits it a borrower of this tag, the same way
+	// a clone of an explicitly named tag is.
+	goldenURI, err := resources.NewTagSnapshotURI(testStorageLocation, resources.GoldenActorAtespace, tag.GetMetadata().GetUid())
+	if err != nil {
+		t.Fatalf("NewTagSnapshotURI: %v", err)
+	}
+	tc.objectStore.PutSnapshot(t, goldenURI, "manifest.json", "memory.zst")
+	tag, err = tc.persistence.UpdateTag(context.Background(), resources.TagRefFromTag(tag), store.PreconditionFrom(tag),
+		func(toUpdate *ateapipb.Tag) error {
+			toUpdate.Status.Snapshot = &ateapipb.ExternalSnapshot{SnapshotUri: goldenURI.String(), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL}
+			return nil
+		})
+	if err != nil {
+		t.Fatalf("record the golden tag's snapshot: %v", err)
+	}
 
 	// Record the golden snapshot on the template's status directly in the
 	// store, as the ActorTemplateReconciler's checkpoint would: there is no
diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go
index 17fc25159e..4aa38b71d2 100644
--- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go
@@ -1392,12 +1392,6 @@ func TestResumeActor_AteletWireRequest(t *testing.T) {
 			tmpl: templateSeed{golden: &ateapipb.ExternalSnapshot{SnapshotUri: goldenURI, ContentScope: dataScope}},
 			want: restoreWant{run: true},
 		},
-		{
-			name:  "06 inherited golden snapshot rejects a malformed URI",
-			actor: actorSeed{externalSnapshot: &ateapipb.ExternalSnapshot{SnapshotUri: malformedURI, ContentScope: fullScope}, tmplUID: "current"},
-			tmpl:  templateSeed{golden: &ateapipb.ExternalSnapshot{SnapshotUri: malformedURI, ContentScope: fullScope}},
-			want:  restoreWant{code: codes.DataLoss},
-		},
 		{
 			name:  "07 template repoint with a late golden still cold-boots",
 			actor: actorSeed{tmplUID: "old-template-uid"},
@@ -1533,11 +1527,6 @@ func TestResumeActor_AteletWireRequest(t *testing.T) {
 				scope:          ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
 			},
 		},
-		{
-			name:  "18 malformed durable snapshot URI fails with DataLoss",
-			actor: actorSeed{externalSnapshot: &ateapipb.ExternalSnapshot{SnapshotUri: malformedURI, ContentScope: fullScope}},
-			want:  restoreWant{code: codes.DataLoss},
-		},
 		{
 			name: "19 Full pause snapshot restores locally as Full",
 			actor: actorSeed{
diff --git a/internal/resources/snapshot.go b/internal/resources/snapshot.go
index c9b553e0ef..33aa55c293 100644
--- a/internal/resources/snapshot.go
+++ b/internal/resources/snapshot.go
@@ -80,6 +80,16 @@ func TagSnapshotOwner(atespace, tagUID string) SnapshotOwner {
 // IsZero reports whether o is the zero SnapshotOwner.
 func (o SnapshotOwner) IsZero() bool { return o == SnapshotOwner{} }
 
+// TagUID returns the UID of the Tag that owns the snapshot, and whether a Tag
+// owns it at all. An Actor holding a tag-owned snapshot is borrowing it, and
+// the Tag may not be deleted out from under it.
+func (o SnapshotOwner) TagUID() (string, bool) {
+	if o.kind != tagsOwnerKind {
+		return "", false
+	}
+	return o.id, true
+}
+
 // Atespace returns the atespace the owner belongs to.
 func (o SnapshotOwner) Atespace() string { return o.atespace }
 

From 29fb9323ed2ab1f467f92949fdb9b388a849fdea Mon Sep 17 00:00:00 2001
From: Luiz Oliveira <luizaoj@google.com>
Date: Fri, 18 Sep 2026 16:46:26 -0400
Subject: [PATCH 6/6] Delete entry from tag_borrowers when tag is deleted

---
 cmd/ateapi/internal/store/atepg/atepg.go      | 14 ++++-
 .../internal/store/storecontract/contract.go  | 58 +++++++++++++++++++
 2 files changed, 71 insertions(+), 1 deletion(-)

diff --git a/cmd/ateapi/internal/store/atepg/atepg.go b/cmd/ateapi/internal/store/atepg/atepg.go
index 86f0b81c9b..dc6dd1b812 100644
--- a/cmd/ateapi/internal/store/atepg/atepg.go
+++ b/cmd/ateapi/internal/store/atepg/atepg.go
@@ -1338,8 +1338,14 @@ func (p *Persistence) UpdateTag(ctx context.Context, tagRef resources.TagRef, pr
 
 func (p *Persistence) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error) {
 	atespace, name := tagRef.Atespace, tagRef.Name
+	tx, err := p.pool.Begin(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("beginning tag delete: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
+
 	var protoBytes []byte
-	if err := p.pool.QueryRow(ctx, `
+	if err := tx.QueryRow(ctx, `
 		DELETE FROM tags
 		WHERE atespace = $1 AND name = $2
 		RETURNING proto`, atespace, name).Scan(&protoBytes); err != nil {
@@ -1352,6 +1358,12 @@ func (p *Persistence) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*
 	if err := unmarshalStored(protoBytes, tag); err != nil {
 		return nil, fmt.Errorf("unmarshaling deleted tag: %w", err)
 	}
+	if _, err := tx.Exec(ctx, `DELETE FROM tag_borrows WHERE tag_uid = $1`, tag.GetMetadata().GetUid()); err != nil {
+		return nil, fmt.Errorf("clearing the borrows of tag %s/%s: %w", atespace, name, err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, fmt.Errorf("committing tag delete: %w", err)
+	}
 	return tag, nil
 }
 
diff --git a/cmd/ateapi/internal/store/storecontract/contract.go b/cmd/ateapi/internal/store/storecontract/contract.go
index 689bf385f9..8949f8e983 100644
--- a/cmd/ateapi/internal/store/storecontract/contract.go
+++ b/cmd/ateapi/internal/store/storecontract/contract.go
@@ -2946,6 +2946,64 @@ func runTagBorrowContractTests(t *testing.T, setup func(t *testing.T) store.Inte
 		}
 	})
 
+	// Nothing else reclaims a borrow row, so one left behind by DeleteTag would
+	// name a Tag that no longer exists for as long as the borrower does.
+	t.Run("TagBorrow_DeleteTag", func(t *testing.T) {
+		s := setup(t)
+		ctx := context.Background()
+		mustCreateAtespace(t, s, testAtespace)
+
+		source, err := s.CreateActor(ctx, newTestSuspendedActor(testAtespace, "actor-source"))
+		if err != nil {
+			t.Fatalf("CreateActor failed: %v", err)
+		}
+		// The store owns the UID of a Tag it creates, so these borrowers point at
+		// the created Tags rather than at the fixed UIDs the other cases use.
+		deletedTag, err := s.CreateTag(ctx, newTestInProgressTag("tag-1", source))
+		if err != nil {
+			t.Fatalf("CreateTag(tag-1) failed: %v", err)
+		}
+		keptTag, err := s.CreateTag(ctx, newTestInProgressTag("tag-2", source))
+		if err != nil {
+			t.Fatalf("CreateTag(tag-2) failed: %v", err)
+		}
+		deletedTagUID, keptTagUID := deletedTag.GetMetadata().GetUid(), keptTag.GetMetadata().GetUid()
+
+		borrower, err := s.CreateActor(ctx, borrowingActor("session-1", deletedTagUID))
+		if err != nil {
+			t.Fatalf("CreateActor borrowing tag-1 failed: %v", err)
+		}
+		keptBorrower, err := s.CreateActor(ctx, borrowingActor("session-2", keptTagUID))
+		if err != nil {
+			t.Fatalf("CreateActor borrowing tag-2 failed: %v", err)
+		}
+		page, err := s.ListTagBorrowers(ctx, deletedTagUID, store.ListOptions{PageSize: 1000})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if diff := cmp.Diff([]string{borrower.GetMetadata().GetUid()}, page.Items); diff != "" {
+			t.Fatalf("borrowers before the tag was deleted (-want +got):\n%s", diff)
+		}
+
+		if _, err := s.DeleteTag(ctx, resources.TagRef{Atespace: testAtespace, Name: "tag-1"}); err != nil {
+			t.Fatalf("DeleteTag failed: %v", err)
+		}
+		page, err = s.ListTagBorrowers(ctx, deletedTagUID, store.ListOptions{PageSize: 1000})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if page.Items != nil {
+			t.Errorf("borrowers after the tag was deleted = %v, want none", page.Items)
+		}
+		page, err = s.ListTagBorrowers(ctx, keptTagUID, store.ListOptions{PageSize: 1000})
+		if err != nil {
+			t.Fatalf("ListTagBorrowers failed: %v", err)
+		}
+		if diff := cmp.Diff([]string{keptBorrower.GetMetadata().GetUid()}, page.Items); diff != "" {
+			t.Errorf("borrowers of the surviving tag (-want +got):\n%s", diff)
+		}
+	})
+
 	t.Run("TagBorrow_ListPagination", func(t *testing.T) {
 		s := setup(t)
 		ctx := context.Background()