diff --git a/agent/cmd/main.go b/agent/cmd/main.go
index 8ce7f15f8..0b5c86d44 100644
--- a/agent/cmd/main.go
+++ b/agent/cmd/main.go
@@ -209,6 +209,13 @@ func main() {
 		protected.Use(authCheck.Middleware, target.Guard(serverUID))
 		mountProtected(protected)
 	})
+	// Backups bind save-off/save-on to the process they prepared. The service
+	// may already route to a replacement pod, which must reject the old UID.
+	r.Route("/v1/pods/{uid}", func(protected chi.Router) {
+		protected.Use(authCheck.Middleware, target.Guard(envOr("GAMEPLANE_POD_UID", "")), middleware.Timeout(30*time.Second))
+		protected.Get("/identity", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
+		quiesce.Mount(protected, rconClient, gameName, quiesceSpec)
+	})
 
 	srv := &http.Server{
 		Addr:              addr,
diff --git a/agent/internal/quiesce/quiesce_test.go b/agent/internal/quiesce/quiesce_test.go
index 777ba9ade..62aca43a4 100644
--- a/agent/internal/quiesce/quiesce_test.go
+++ b/agent/internal/quiesce/quiesce_test.go
@@ -12,6 +12,7 @@ import (
 	"github.com/go-chi/chi/v5"
 
 	"github.com/GameplanePanel/gameplane/agent/internal/caps"
+	"github.com/GameplanePanel/gameplane/agent/internal/target"
 )
 
 type fakeRcon struct {
@@ -86,6 +87,33 @@ func newTestRouter(rc Rcon, spec *caps.Quiesce) *chi.Mux {
 	return r
 }
 
+func TestPodBoundRoutesRejectReplacementBeforeRCON(t *testing.T) {
+	for _, actual := range []string{"original", "replacement", ""} {
+		t.Run("pod="+actual, func(t *testing.T) {
+			rc := &fakeRcon{}
+			r := chi.NewRouter()
+			r.Route("/v1/pods/{uid}", func(bound chi.Router) {
+				bound.Use(target.Guard(actual))
+				Mount(bound, rc, "test", minecraftQuiesceSpec())
+			})
+			for _, path := range []string{"quiesce", "unquiesce"} {
+				response := httptest.NewRecorder()
+				r.ServeHTTP(response, httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/v1/pods/original/"+path, nil))
+				if actual == "original" {
+					if response.Code != http.StatusOK {
+						t.Fatalf("matching process rejected: %d", response.Code)
+					}
+				} else if response.Code != http.StatusNotFound {
+					t.Fatalf("wrong process admitted: %d", response.Code)
+				}
+			}
+			if (len(rc.calls()) > 0) != (actual == "original") {
+				t.Fatalf("commands reached wrong process: %v", rc.calls())
+			}
+		})
+	}
+}
+
 func decodeResponse(t *testing.T, body []byte) response {
 	t.Helper()
 	var got response
diff --git a/agent/internal/target/target.go b/agent/internal/target/target.go
index 6a1300631..3e79a3944 100644
--- a/agent/internal/target/target.go
+++ b/agent/internal/target/target.go
@@ -1,4 +1,4 @@
-// Package target binds versioned agent routes to an immutable GameServer UID.
+// Package target binds versioned agent routes to an immutable resource UID.
 package target
 
 import (
@@ -7,12 +7,13 @@ import (
 	"github.com/go-chi/chi/v5"
 )
 
-// Guard rejects a route for any server other than the one that started this
-// agent. Legacy routes stay available for local callers during rolling upgrades.
-func Guard(serverUID string) func(http.Handler) http.Handler {
+// Guard rejects a route for any resource other than the one that started this
+// agent (GameServer or Pod, according to the route). Legacy routes stay available
+// for local callers during rolling upgrades.
+func Guard(resourceUID string) func(http.Handler) http.Handler {
 	return func(next http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-			if serverUID == "" || chi.URLParam(req, "uid") != serverUID {
+			if resourceUID == "" || chi.URLParam(req, "uid") != resourceUID {
 				http.NotFound(w, req)
 				return
 			}
diff --git a/agent/specs.md b/agent/specs.md
index 13d03c989..1f26ec777 100644
--- a/agent/specs.md
+++ b/agent/specs.md
@@ -177,6 +177,8 @@ Resource usage env vars (set by the operator):
 | `/players/banned` | GET | Currently banned players; response: `[ { name, reason?, source? } ]`; 502 if RCON unavailable |
 | `/quiesce` | POST | Pause auto-saves before snapshot; response: `{ quiesced, reason? }` (boolean, reason optional); unsupported games return `quiesced: false` with a reason |
 | `/unquiesce` | POST | Resume auto-saves after snapshot; response: `{ quiesced, reason? }` |
+| `/v1/pods/{uid}/identity` | GET | Operator preflight: 204 only when `{uid}` matches `GAMEPLANE_POD_UID`; never mutates game state |
+| `/v1/pods/{uid}/quiesce`, `/v1/pods/{uid}/unquiesce` | POST | Pod-bound backup RPCs with the same response contract as the legacy routes; reject missing/mismatched Pod UID with 404 before RCON |
 | `/lifecycle/stop` | POST | Run stop sequence before scale-to-zero; RCON connection drop is the expected outcome |
 | `/actions/run` | POST | Execute module-declared operator action; request: `{ id, params }`; response: `{ ok, raw? }` |
 | `/status` | GET | Live game metrics (module-declared); response: bare JSON array `[ { id, displayName?, value, unit? } ]` |
@@ -209,6 +211,7 @@ All endpoints on the `--addr` control mux, except `/healthz`, return `401 Unauth
 - **Extracted mod files stay world-readable**: `mods.moduleFileMode` is `0o644`, not a tighter `0o600`, because the mods volume is shared with the game container, which runs as whatever uid its image needs — a different uid than the agent's own. At `0o600` the game process couldn't read its own mods. This is the rationale behind the `.golangci.yml` gosec G302 exclusion scoped to `agent/internal/mods/mods.go`.
 - **Resource usage is in-pod**: The `usage` package reads from `/proc` or cgroups; no external metrics pipeline required. Cgroup mode is a fallback for older clusters; proc mode (default in production) requires the operator to set `ShareProcessNamespace: true`.
 - **Module capabilities drive behavior**: Every game-specific handler (players, quiesce, lifecycle, status, actions) reads its config from `--capabilities` (JSON unmarshaled into `caps.Spec`). New games require no agent code change.
+- **Backup process identity**: The operator injects `GAMEPLANE_POD_UID` from `metadata.uid` through the DownwardAPI. Pod-bound backup routes require the existing mTLS authentication and check this UID at the final network hop, including when the Service resolves to a replacement pod. Missing UID fails closed. Deploy the updated agent image and allow pod rollout when upgrading the operator; older agents return 404 and the operator does not fall back to unbound routes. Legacy endpoints remain available to existing callers during rollout.
 - **RCON connection errors are graceful**: A lost RCON connection does not crash the agent. `console`, `players`, `quiesce`, and `lifecycle` handlers catch connection errors and return appropriate HTTP status (e.g., `502 Bad Gateway`).
 - **Source RCON sends complete frames**: Each AUTH or command frame is submitted in one socket write, with its existing byte-size limit, request ID, type, body, and two NUL terminators preserved. This avoids separate header-only writes rejected by vanilla Minecraft's RCON reader. A short write is an error and drops the connection without replaying the command; response grace and healthy connection reuse remain unchanged.
 - **Log streams are tail-only**: The `logs` package does not support random-access reads. It streams from the current end (live mode) or from file start (backlog mode); clients must handle partial output and reconnection.
diff --git a/i18n/locales/en/app.json b/i18n/locales/en/app.json
index eeaeda262..4b17d4180 100644
--- a/i18n/locales/en/app.json
+++ b/i18n/locales/en/app.json
@@ -607,6 +607,10 @@
     "defaultMessage": "The server could not be resumed after the backup. Its world may still have auto-save turned off.",
     "description": "Unquiesced condition when the unfreeze failed."
   },
+  "condition.unquiesced.sourceprocessgone": {
+    "defaultMessage": "The original game process has exited. No resume command was needed.",
+    "description": "Unquiesced condition when the original source pod no longer runs."
+  },
   "condition.unquiesced.unquiesced": {
     "defaultMessage": "The server was resumed after the backup.",
     "description": "Unquiesced condition when the world was unfrozen."
diff --git a/operator/hack/condition-ids/main.go b/operator/hack/condition-ids/main.go
index ef0c4fa69..1ecb4dc95 100644
--- a/operator/hack/condition-ids/main.go
+++ b/operator/hack/condition-ids/main.go
@@ -142,6 +142,7 @@ var rows = []row{
 	{types: []string{"Completed"}, reason: "Pending", english: "The backup is waiting to start.", description: "Completed condition on a Backup that has not started (dynamic: reason is the phase).", dynamic: true},
 	{types: []string{"Completed"}, reason: "Running", english: "The backup is running.", description: "Completed condition on a Backup in progress (dynamic: reason is the phase).", dynamic: true},
 	{types: []string{"Unquiesced"}, reason: "Unquiesced", english: "The server was resumed after the backup.", description: "Unquiesced condition when the world was unfrozen."},
+	{types: []string{"Unquiesced"}, reason: "SourceProcessGone", english: "The original game process has exited. No resume command was needed.", description: "Unquiesced condition when the original source pod no longer runs."},
 	{types: []string{"Unquiesced"}, reason: "Failed", english: "The server could not be resumed after the backup. Its world may still have auto-save turned off.", description: "Unquiesced condition when the unfreeze failed."},
 	{types: []string{"SnapshotForgotten"}, reason: "Forgotten", english: "The snapshot was removed from the backup repository.", description: "SnapshotForgotten condition after a deleted Backup's snapshot was forgotten."},
 	{types: []string{"SnapshotForgotten"}, reason: "ForgetAbandoned", english: "The snapshot was not removed from the backup repository when the backup was deleted.", description: "SnapshotForgotten condition when the forget step was abandoned."},
diff --git a/operator/internal/agent/client.go b/operator/internal/agent/client.go
index d606c3c75..2b1c06923 100644
--- a/operator/internal/agent/client.go
+++ b/operator/internal/agent/client.go
@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"os"
 	"time"
 )
@@ -119,6 +120,47 @@ func (c *Client) Unquiesce(ctx context.Context, namespace, server string) error
 	return c.callQuiesce(ctx, namespace, server, "/unquiesce", false)
 }
 
+// QuiesceEnabled reports whether this client can perform quiesce operations.
+func (c *Client) QuiesceEnabled() bool { return c != nil && !c.Disabled }
+
+// CheckPod verifies that the responding agent belongs to the selected Pod.
+// Older agents return 404; callers must not fall back to unbound mutations.
+func (c *Client) CheckPod(ctx context.Context, namespace, server, podUID string) error {
+	if c.Disabled || podUID == "" {
+		return fmt.Errorf("agent: pod identity check requires an enabled client and pod UID")
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, agentURL(namespace, server, "/v1/pods/"+url.PathEscape(podUID)+"/identity"), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("agent: pod identity check: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode != http.StatusNoContent {
+		return fmt.Errorf("agent: pod identity check: status %d", resp.StatusCode)
+	}
+	return nil
+}
+
+// QuiescePod disables saving only on the selected Pod's agent.
+func (c *Client) QuiescePod(ctx context.Context, namespace, server, podUID string) error {
+	return c.callPodQuiesce(ctx, namespace, server, podUID, "/quiesce", true)
+}
+
+// UnquiescePod restores saving only on the selected Pod's agent.
+func (c *Client) UnquiescePod(ctx context.Context, namespace, server, podUID string) error {
+	return c.callPodQuiesce(ctx, namespace, server, podUID, "/unquiesce", false)
+}
+
+func (c *Client) callPodQuiesce(ctx context.Context, namespace, server, podUID, path string, expectQuiesced bool) error {
+	if c.Disabled || podUID == "" {
+		return fmt.Errorf("agent: pod-bound quiesce requires an enabled client and pod UID")
+	}
+	return c.callQuiesce(ctx, namespace, server, "/v1/pods/"+url.PathEscape(podUID)+path, expectQuiesced)
+}
+
 // expectQuiesced is what the response should report on a successful
 // completion: true after /quiesce, false after /unquiesce. A mismatch
 // is a protocol bug worth surfacing.
diff --git a/operator/internal/agent/client_test.go b/operator/internal/agent/client_test.go
index c3866488e..7d04dc9d1 100644
--- a/operator/internal/agent/client_test.go
+++ b/operator/internal/agent/client_test.go
@@ -122,6 +122,46 @@ func TestAgentURL(t *testing.T) {
 	}
 }
 
+func TestPodBoundQuiesceNeverFallsBackToLegacyRoute(t *testing.T) {
+	for _, status := range []int{http.StatusOK, http.StatusNotFound} {
+		t.Run(http.StatusText(status), func(t *testing.T) {
+			paths := []string{}
+			c, closeServer := newMockClient(t, func(w http.ResponseWriter, r *http.Request) {
+				paths = append(paths, r.URL.Path)
+				w.WriteHeader(status)
+				_, _ = w.Write([]byte(`{"quiesced":true}`))
+			})
+			defer closeServer()
+			err := c.QuiescePod(t.Context(), "ns", "server", "original-pod")
+			if (err == nil) != (status == http.StatusOK) {
+				t.Fatalf("status=%d error=%v", status, err)
+			}
+			if len(paths) != 1 || paths[0] != "/v1/pods/original-pod/quiesce" {
+				t.Fatalf("unbound mutation/fallback: %v", paths)
+			}
+		})
+	}
+}
+
+func TestPodBoundUnquiesceAndDisabledClient(t *testing.T) {
+	c, closeServer := newMockClient(t, func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/pods/original-pod/unquiesce" {
+			t.Errorf("wrong cleanup target: %s", r.URL.Path)
+		}
+		_, _ = w.Write([]byte(`{"quiesced":false}`))
+	})
+	defer closeServer()
+	if err := c.UnquiescePod(t.Context(), "ns", "server", "original-pod"); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.QuiescePod(t.Context(), "ns", "server", ""); err == nil {
+		t.Fatal("empty pod UID admitted")
+	}
+	if err := (&Client{Disabled: true}).QuiescePod(t.Context(), "ns", "server", "pod"); err == nil {
+		t.Fatal("disabled client reported identity-bound quiesce")
+	}
+}
+
 func TestQuiesce_DisabledNoOp(t *testing.T) {
 	c := &Client{Disabled: true}
 	if err := c.Quiesce(context.Background(), "ns", "srv"); err != nil {
@@ -257,6 +297,32 @@ func TestUnquiesce_HappyPath(t *testing.T) {
 	}
 }
 
+func TestCheckPodRequiresIdentityEndpoint(t *testing.T) {
+	for _, status := range []int{http.StatusNoContent, http.StatusNotFound, http.StatusOK, http.StatusServiceUnavailable} {
+		t.Run(http.StatusText(status), func(t *testing.T) {
+			calls := 0
+			c, cleanup := newMockClient(t, func(w http.ResponseWriter, req *http.Request) {
+				calls++
+				if req.Method != http.MethodGet || req.URL.Path != "/v1/pods/original/identity" {
+					t.Errorf("unexpected preflight request: %s %s", req.Method, req.URL.Path)
+				}
+				w.WriteHeader(status)
+			})
+			defer cleanup()
+			err := c.CheckPod(t.Context(), "ns", "server", "original")
+			if (err == nil) != (status == http.StatusNoContent) || calls != 1 {
+				t.Fatalf("identity result = %v, calls = %d", err, calls)
+			}
+		})
+	}
+	if err := (&Client{Disabled: true}).CheckPod(t.Context(), "ns", "server", "original"); err == nil {
+		t.Fatal("disabled client confirmed identity")
+	}
+	if err := (&Client{}).CheckPod(t.Context(), "ns", "server", ""); err == nil {
+		t.Fatal("empty pod UID confirmed identity")
+	}
+}
+
 func TestQuiesce_TransportError(t *testing.T) {
 	// Use a base transport that always errors.
 	c := &Client{
diff --git a/operator/internal/controller/backup_controller.go b/operator/internal/controller/backup_controller.go
index bf50ec773..cefa931b7 100644
--- a/operator/internal/controller/backup_controller.go
+++ b/operator/internal/controller/backup_controller.go
@@ -386,15 +386,47 @@ func (r *BackupReconciler) reconcile(ctx context.Context, req ctrl.Request) (ctr
 // requeues; ErrUnsupported is recorded as "attempted but unsupported"
 // so the matching unquiesce step is skipped on terminal phase.
 func (r *BackupReconciler) maybeQuiesce(ctx context.Context, b *gameplanev1alpha1.Backup) error {
+	preparation, err := parseBackupPreparation(b)
+	if err != nil {
+		return err
+	}
+	if preparation != nil && preparation.Outcome == "already-stopped" {
+		return nil
+	}
 	if !b.Spec.Quiesce {
 		return nil
 	}
-	if r.AgentClient == nil {
+	if !agentQuiesceAvailable(r.AgentClient) {
+		if preparation != nil && preparation.Outcome == "live" && b.Annotations[annoQuiesceAttempted] != "true" && b.Annotations[annoQuiesceAttempted] != "unsupported" {
+			err := fmt.Errorf("agent quiesce client was disabled after live preparation")
+			if _, ferr := r.fail(ctx, b, err.Error()); ferr != nil {
+				return ferr
+			}
+			return err
+		}
 		return nil
 	}
 	if state := b.Annotations[annoQuiesceAttempted]; state != "" && state != "pending" {
 		return nil
 	}
+	// Verify the versioned endpoint before creating the first cleanup debt.
+	// Old agents cannot mutate through it and must not pin an owner that
+	// would prevent their own upgrade. The POST still checks the UID again.
+	if preparation != nil && b.Annotations[annoQuiesceAttempted] == "" {
+		bound, ok := r.AgentClient.(podAgentQuiescer)
+		if !ok {
+			err = fmt.Errorf("agent client does not support pod-bound quiesce")
+		} else {
+			err = bound.CheckPod(ctx, b.Namespace, b.Spec.ServerRef.Name, string(preparation.PodUID))
+		}
+		if err != nil {
+			_, ferr := r.fail(ctx, b, fmt.Sprintf("quiesce identity check failed: %v", err))
+			if ferr != nil {
+				return ferr
+			}
+			return err
+		}
+	}
 	// Persist cleanup intent BEFORE save-off: a crash or a partial agent failure
 	// must still owe save-on even when quiesced-at was never recorded.
 	if err := r.setLifecycleUnquiesceIntent(ctx, b, true); err != nil {
@@ -405,7 +437,11 @@ func (r *BackupReconciler) maybeQuiesce(ctx context.Context, b *gameplanev1alpha
 		return err
 	}
 
-	err := r.AgentClient.Quiesce(ctx, b.Namespace, b.Spec.ServerRef.Name)
+	var podUID types.UID
+	if preparation != nil {
+		podUID = preparation.PodUID
+	}
+	err = callBackupQuiesce(ctx, r.AgentClient, b.Namespace, b.Spec.ServerRef.Name, podUID, false)
 	switch {
 	case err == nil:
 		now := time.Now().UTC().Format(time.RFC3339)
@@ -629,8 +665,8 @@ func (r *BackupReconciler) runUnquiesce(ctx context.Context, b *gameplanev1alpha
 		}
 		return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
 	}
-	// When an unquiesce was actually performed, record success so the
-	// "world might be frozen" signal clears.
+	// Confirmed save-on or confirmed original-process exit clears the
+	// "world might be frozen" signal, with distinct condition reasons.
 	if _, ok := b.Annotations[annoUnquiescedAt]; ok {
 		if cerr := r.setUnquiescedCondition(ctx, b, true, ""); cerr != nil {
 			return ctrl.Result{}, cerr
@@ -649,6 +685,10 @@ func (r *BackupReconciler) setUnquiescedCondition(
 	if ok {
 		cond.Status = metav1.ConditionTrue
 		cond.Reason = "Unquiesced"
+		if b.Annotations[backupCleanupOutcomeAnnotation] == "source-gone" {
+			cond.Reason = "SourceProcessGone"
+			cond.Message = "original process no longer exists; no unquiesce RPC was sent"
+		}
 	} else {
 		cond.Status = metav1.ConditionFalse
 		cond.Reason = "Failed"
@@ -740,49 +780,29 @@ func (r *BackupReconciler) finalizeUnquiesce(ctx context.Context, b *gameplanev1
 	return ctrl.Result{}, nil
 }
 
-// unquiesceTargetGone reports whether there is nothing left to unquiesce:
-// the target GameServer no longer exists or is itself being deleted, or its
-// pod no longer exists. A GameServer can be gone-together-with-its-Backup
-// under an auto BackupSchedule's owner chain, and a pod alone can be gone if
-// it was deleted independently of the GameServer; either way a fresh pod
-// starts with auto-save on, so retrying the unquiesce would never succeed
-// and would only block this Backup's deletion.
+// unquiesceTargetGone requires evidence that the original process has exited.
+// A deleted/deleting GameServer alone is insufficient: its pod can still run.
 func (r *BackupReconciler) unquiesceTargetGone(ctx context.Context, b *gameplanev1alpha1.Backup) (bool, error) {
 	targetName := b.Spec.ServerRef.Name
-	var targetUID types.UID
 	if value := b.Annotations[backupTargetAnnotation]; value != "" {
 		identity, err := parseBackupIdentity(value)
 		if err != nil {
 			return false, err
 		}
-		targetName, targetUID = identity.Name, identity.UID
+		targetName = identity.Name
 	}
 	if targetName == "" {
 		return true, nil
 	}
-	gs := &gameplanev1alpha1.GameServer{}
-	key := types.NamespacedName{Namespace: b.Namespace, Name: targetName}
-	switch err := r.Get(ctx, key, gs); {
-	case apierrors.IsNotFound(err):
-		return true, nil
-	case err != nil:
-		return false, fmt.Errorf("get gameserver %s to check unquiesce target: %w", key.Name, err)
-	case gs.DeletionTimestamp != nil:
-		return true, nil
-	}
-	if targetUID != "" && gs.UID != targetUID {
-		return true, nil
+	preparation, err := parseBackupPreparation(b)
+	if err != nil {
+		return false, err
 	}
-
-	var pod corev1.Pod
-	podKey := types.NamespacedName{Namespace: b.Namespace, Name: gs.Name + "-0"}
-	switch err := r.Get(ctx, podKey, &pod); {
-	case apierrors.IsNotFound(err):
-		return true, nil
-	case err != nil:
-		return false, fmt.Errorf("get pod %s to check unquiesce target: %w", podKey.Name, err)
+	var podUID types.UID
+	if preparation != nil {
+		podUID = preparation.PodUID
 	}
-	return false, nil
+	return backupQuiesceProcessGone(ctx, r.backupAPIReader(), b.Namespace, targetName, podUID)
 }
 
 // unquiesceRetryExpired records (on first call) when finalizeDelete's
@@ -870,10 +890,35 @@ func (r *BackupReconciler) maybeUnquiesce(ctx context.Context, b *gameplanev1alp
 			return fmt.Errorf("legacy quiesce cleanup is blocked by a protected backup")
 		}
 	}
-	if r.AgentClient == nil {
+	if targetName != "" {
+		gone, err := r.unquiesceTargetGone(ctx, b)
+		if err != nil {
+			return err
+		}
+		if gone {
+			patchBackupAnnotations(b, map[string]string{backupCleanupOutcomeAnnotation: "source-gone"})
+			return r.recordBackupUnquiesced(ctx, b)
+		}
+	}
+	if !agentQuiesceAvailable(r.AgentClient) {
 		return fmt.Errorf("cannot release quiesced world without an agent client")
 	}
-	if err := r.AgentClient.Unquiesce(ctx, b.Namespace, targetName); err != nil {
+	preparation, err := parseBackupPreparation(b)
+	if err != nil {
+		return err
+	}
+	var podUID types.UID
+	if preparation != nil {
+		podUID = preparation.PodUID
+	}
+	if err := callBackupQuiesce(ctx, r.AgentClient, b.Namespace, targetName, podUID, true); err != nil {
+		return err
+	}
+	return r.recordBackupUnquiesced(ctx, b)
+}
+
+func (r *BackupReconciler) recordBackupUnquiesced(ctx context.Context, b *gameplanev1alpha1.Backup) error {
+	if err := r.setLifecycleUnquiesceIntent(ctx, b, false); err != nil {
 		return err
 	}
 	patchBackupAnnotations(b, map[string]string{
diff --git a/operator/internal/controller/backup_envtest_test.go b/operator/internal/controller/backup_envtest_test.go
index 9b037564a..f577563a6 100644
--- a/operator/internal/controller/backup_envtest_test.go
+++ b/operator/internal/controller/backup_envtest_test.go
@@ -8,9 +8,11 @@ import (
 	"testing"
 	"time"
 
+	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1"
 )
@@ -379,6 +381,20 @@ func TestBackup_QuiesceAnnotationSetBeforeJob(t *testing.T) {
 	fa := &fakeAgent{}
 	startMgr(t, ns, withBackupReconciler(backupReconcilerOpts{agent: fa}))
 	seedGameServer(t, ns, "smp")
+	// envtest has no StatefulSet controller or kubelet: provide the observed
+	// running source whose immutable identity is bound before the agent RPC.
+	var workload appsv1.StatefulSet
+	if err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: ns, Name: "smp"}, &workload); err != nil {
+		t.Fatal(err)
+	}
+	pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "smp-0", Namespace: ns, OwnerReferences: []metav1.OwnerReference{{APIVersion: "apps/v1", Kind: "StatefulSet", Name: workload.Name, UID: workload.UID, Controller: ownerBoolPtr(true)}}}, Spec: workload.Spec.Template.Spec}
+	if err := k8sClient.Create(context.Background(), pod); err != nil {
+		t.Fatal(err)
+	}
+	pod.Status.Phase = corev1.PodRunning
+	if err := k8sClient.Status().Update(context.Background(), pod); err != nil {
+		t.Fatal(err)
+	}
 
 	if err := k8sClient.Create(context.Background(), buildResticRepoSecret(ns, "repo")); err != nil {
 		t.Fatalf("create secret: %v", err)
diff --git a/operator/internal/controller/backup_guard.go b/operator/internal/controller/backup_guard.go
index cee37734e..2b6e4d4f2 100644
--- a/operator/internal/controller/backup_guard.go
+++ b/operator/internal/controller/backup_guard.go
@@ -119,11 +119,19 @@ func (r *BackupReconciler) recoverBackupTarget(ctx context.Context, b *gameplane
 			return false, err
 		}
 	}
-	if r.AgentClient == nil {
-		return false, fmt.Errorf("cannot recover orphaned quiesce without an agent client")
+	// Legacy owners have no pod UID. Retire them only on positive absence or
+	// terminal-pod proof; never bind an old RPC to a newly observed identity.
+	gone, err := backupQuiesceProcessGone(ctx, r.backupAPIReader(), b.Namespace, b.Spec.ServerRef.Name, "")
+	if err != nil {
+		return false, err
 	}
-	if err := r.AgentClient.Unquiesce(ctx, b.Namespace, b.Spec.ServerRef.Name); err != nil {
-		return false, fmt.Errorf("recover orphaned quiesce: %w", err)
+	if !gone {
+		if !agentQuiesceAvailable(r.AgentClient) {
+			return false, fmt.Errorf("cannot recover orphaned quiesce without an agent client")
+		}
+		if err := r.AgentClient.Unquiesce(ctx, b.Namespace, b.Spec.ServerRef.Name); err != nil {
+			return false, fmt.Errorf("recover orphaned quiesce: %w", err)
+		}
 	}
 	delete(b.Annotations, backupRecoveryAnnotation)
 	delete(b.Annotations, annoQuiesceAttempted)
@@ -267,7 +275,28 @@ func (r *BackupReconciler) prepareBackupQuiesce(ctx context.Context, b *gameplan
 	if err != nil || !fenced {
 		return ctrl.Result{RequeueAfter: 5 * time.Second}, err
 	}
-	return ctrl.Result{}, r.maybeQuiesce(ctx, b)
+	ready, err := r.prepareBackupSource(ctx, b, gs)
+	if errors.Is(err, errBackupSourceChanged) {
+		_, err = r.fail(ctx, b, err.Error())
+		return ctrl.Result{RequeueAfter: time.Nanosecond}, err
+	}
+	if err != nil || !ready {
+		return ctrl.Result{RequeueAfter: 5 * time.Second}, err
+	}
+	if err := r.maybeQuiesce(ctx, b); err != nil {
+		return ctrl.Result{}, err
+	}
+	// The Service RPC may overlap a pod replacement. Never start a copy on a
+	// different process from the one pinned before the request.
+	ready, err = r.prepareBackupSource(ctx, b, gs)
+	if errors.Is(err, errBackupSourceChanged) {
+		_, err = r.fail(ctx, b, err.Error())
+		return ctrl.Result{RequeueAfter: time.Nanosecond}, err
+	}
+	if err != nil || !ready {
+		return ctrl.Result{RequeueAfter: 5 * time.Second}, err
+	}
+	return ctrl.Result{}, nil
 }
 
 // Return only the originally bound GameServer. A foreign owner means this
diff --git a/operator/internal/controller/backup_guard_test.go b/operator/internal/controller/backup_guard_test.go
index 8b40e1104..b6d2ccb2d 100644
--- a/operator/internal/controller/backup_guard_test.go
+++ b/operator/internal/controller/backup_guard_test.go
@@ -28,13 +28,30 @@ import (
 type guardQuiescer struct {
 	quiesced, unquiesced int
 	quiesceErr           error
+	unquiesceErr         error
+	checkErr             error
+	afterQuiesce         func()
 }
 
 func (q *guardQuiescer) Quiesce(context.Context, string, string) error {
 	q.quiesced++
 	return q.quiesceErr
 }
-func (q *guardQuiescer) Unquiesce(context.Context, string, string) error { q.unquiesced++; return nil }
+func (q *guardQuiescer) Unquiesce(context.Context, string, string) error {
+	q.unquiesced++
+	return q.unquiesceErr
+}
+func (q *guardQuiescer) CheckPod(context.Context, string, string, string) error { return q.checkErr }
+func (q *guardQuiescer) QuiescePod(ctx context.Context, namespace, server, _ string) error {
+	err := q.Quiesce(ctx, namespace, server)
+	if q.afterQuiesce != nil {
+		q.afterQuiesce()
+	}
+	return err
+}
+func (q *guardQuiescer) UnquiescePod(ctx context.Context, namespace, server, _ string) error {
+	return q.Unquiesce(ctx, namespace, server)
+}
 
 func guardBackup(name string) *gameplanev1alpha1.Backup {
 	return &gameplanev1alpha1.Backup{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", UID: types.UID(name + "-uid"), Finalizers: []string{gameplanev1alpha1.BackupFinalizer, gameplanev1alpha1.BackupSnapshotFinalizer}}, Spec: gameplanev1alpha1.BackupSpec{ServerRef: gameplanev1alpha1.LocalObjectRef{Name: "server"}, RepoRef: &gameplanev1alpha1.SecretKeySelector{Name: "repo"}, Quiesce: true}}
@@ -52,6 +69,7 @@ func guardFixture(t *testing.T, backups ...*gameplanev1alpha1.Backup) (*BackupRe
 	objects := []client.Object{&gameplanev1alpha1.GameServer{ObjectMeta: metav1.ObjectMeta{Name: "server", Namespace: "ns", UID: "server-uid"}, Spec: gameplanev1alpha1.GameServerSpec{TemplateRef: gameplanev1alpha1.GameTemplateRef{Name: "template"}}}, &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "repo", Namespace: "ns"}, Data: map[string][]byte{"repo": []byte("repo"), "password": []byte("password")}}}
 	one := int32(1)
 	objects = append(objects, &appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: "server", Namespace: "ns", UID: "workload-uid", OwnerReferences: []metav1.OwnerReference{{APIVersion: gameplanev1alpha1.GroupVersion.String(), Kind: "GameServer", Name: "server", UID: "server-uid", Controller: ownerBoolPtr(true)}}}, Spec: appsv1.StatefulSetSpec{Replicas: &one}})
+	objects = append(objects, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "server-0", Namespace: "ns", UID: "game-pod-uid", OwnerReferences: []metav1.OwnerReference{{APIVersion: "apps/v1", Kind: "StatefulSet", Name: "server", UID: "workload-uid", Controller: ownerBoolPtr(true)}}}, Status: corev1.PodStatus{Phase: corev1.PodRunning}})
 	for _, b := range backups {
 		objects = append(objects, b)
 	}
diff --git a/operator/internal/controller/backup_preparation.go b/operator/internal/controller/backup_preparation.go
new file mode 100644
index 000000000..4f682ab53
--- /dev/null
+++ b/operator/internal/controller/backup_preparation.go
@@ -0,0 +1,234 @@
+package controller
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1"
+)
+
+const (
+	backupPreparationAnnotation    = "backup.gameplane.local/preparation"
+	backupCleanupOutcomeAnnotation = "backup.gameplane.local/cleanup-outcome"
+)
+
+var errBackupSourceChanged = errors.New("backup source changed after preparation")
+
+type podAgentQuiescer interface {
+	CheckPod(context.Context, string, string, string) error
+	QuiescePod(context.Context, string, string, string) error
+	UnquiescePod(context.Context, string, string, string) error
+}
+
+func agentQuiesceAvailable(q AgentQuiescer) bool {
+	if q == nil {
+		return false
+	}
+	if configured, ok := q.(interface{ QuiesceEnabled() bool }); ok {
+		return configured.QuiesceEnabled()
+	}
+	return true
+}
+
+func callBackupQuiesce(ctx context.Context, q AgentQuiescer, namespace, server string, podUID types.UID, undo bool) error {
+	if !agentQuiesceAvailable(q) {
+		return fmt.Errorf("agent quiesce client is not configured")
+	}
+	if podUID != "" {
+		bound, ok := q.(podAgentQuiescer)
+		if !ok {
+			return fmt.Errorf("agent client does not support pod-bound quiesce")
+		}
+		if undo {
+			return bound.UnquiescePod(ctx, namespace, server, string(podUID))
+		}
+		return bound.QuiescePod(ctx, namespace, server, string(podUID))
+	}
+	if undo {
+		return q.Unquiesce(ctx, namespace, server)
+	}
+	return q.Quiesce(ctx, namespace, server)
+}
+
+// Source identity is immutable for one copy. RPC outcome remains in the
+// quiesce annotations, so a stopped source is never described as quiesced.
+type backupPreparation struct {
+	Outcome            string    `json:"outcome"`
+	TargetUID          types.UID `json:"targetUID"`
+	WorkloadUID        types.UID `json:"workloadUID"`
+	WorkloadGeneration int64     `json:"workloadGeneration"`
+	PodUID             types.UID `json:"podUID,omitempty"`
+}
+
+func parseBackupPreparation(b *gameplanev1alpha1.Backup) (*backupPreparation, error) {
+	value := b.Annotations[backupPreparationAnnotation]
+	if value == "" {
+		return nil, nil
+	}
+	var p backupPreparation
+	if err := json.Unmarshal([]byte(value), &p); err != nil {
+		return nil, fmt.Errorf("decode backup preparation: %w", err)
+	}
+	if p.TargetUID == "" || p.WorkloadUID == "" || (p.Outcome != "live" && p.Outcome != "already-stopped") || (p.Outcome == "live" && p.PodUID == "") || (p.Outcome == "already-stopped" && p.PodUID != "") {
+		return nil, fmt.Errorf("invalid backup preparation identity")
+	}
+	return &p, nil
+}
+
+// Call only after the shared owner and StatefulSet fence are held. Readiness
+// and desired suspend/idle state are not evidence that writers have exited.
+func (r *BackupReconciler) prepareBackupSource(ctx context.Context, b *gameplanev1alpha1.Backup, gs *gameplanev1alpha1.GameServer) (bool, error) {
+	if !b.Spec.Quiesce {
+		return true, nil
+	}
+	p, err := parseBackupPreparation(b)
+	if err != nil {
+		return false, err
+	}
+	if p == nil && !agentQuiesceAvailable(r.AgentClient) {
+		return true, nil
+	}
+	// Pre-upgrade attempts have no source identity. Do not invent a new one
+	// after their RPC; preserve their existing cleanup contract.
+	if p == nil && b.Annotations[annoQuiesceAttempted] != "" {
+		return true, nil
+	}
+	var ss appsv1.StatefulSet
+	if err := r.backupAPIReader().Get(ctx, client.ObjectKeyFromObject(gs), &ss); err != nil {
+		if apierrors.IsNotFound(err) && p != nil {
+			return false, fmt.Errorf("%w: workload absent", errBackupSourceChanged)
+		}
+		return false, client.IgnoreNotFound(err)
+	}
+	if p != nil && (p.TargetUID != gs.UID || p.WorkloadUID != ss.UID || p.WorkloadGeneration != ss.Generation) {
+		return false, fmt.Errorf("%w: workload identity", errBackupSourceChanged)
+	}
+	if !metav1.IsControlledBy(&ss, gs) {
+		return false, fmt.Errorf("backup source workload is not owned by the bound GameServer")
+	}
+	if ss.Spec.Replicas == nil || ss.Status.ObservedGeneration < ss.Generation {
+		return false, nil
+	}
+	if p != nil && ((p.Outcome == "already-stopped") != (*ss.Spec.Replicas == 0)) {
+		return false, fmt.Errorf("%w: desired replicas", errBackupSourceChanged)
+	}
+	var candidate backupPreparation
+	candidate.TargetUID, candidate.WorkloadUID, candidate.WorkloadGeneration = gs.UID, ss.UID, ss.Generation
+	if *ss.Spec.Replicas == 0 {
+		if ss.Status.Replicas != 0 {
+			if p != nil {
+				return false, fmt.Errorf("%w: stopped workload has replicas", errBackupSourceChanged)
+			}
+			return false, nil
+		}
+		drained, err := r.backupSourceDrained(ctx, b, gs, &ss)
+		if err != nil || !drained {
+			if err == nil && p != nil {
+				return false, fmt.Errorf("%w: stopped source acquired a writer", errBackupSourceChanged)
+			}
+			return false, err
+		}
+		candidate.Outcome = "already-stopped"
+	} else {
+		var pod corev1.Pod
+		err := r.backupAPIReader().Get(ctx, client.ObjectKey{Namespace: gs.Namespace, Name: gs.Name + "-0"}, &pod)
+		if apierrors.IsNotFound(err) {
+			if p != nil {
+				return false, fmt.Errorf("%w: original pod absent", errBackupSourceChanged)
+			}
+			return false, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		if p != nil && (p.PodUID != pod.UID || !pod.DeletionTimestamp.IsZero() || pod.Status.Phase != corev1.PodRunning) {
+			return false, fmt.Errorf("%w: original process no longer running", errBackupSourceChanged)
+		}
+		if !metav1.IsControlledBy(&pod, &ss) || pod.UID == "" {
+			return false, fmt.Errorf("backup source pod is not owned by the bound workload")
+		}
+		if !pod.DeletionTimestamp.IsZero() || pod.Status.Phase != corev1.PodRunning {
+			return false, nil
+		}
+		candidate.Outcome, candidate.PodUID = "live", pod.UID
+	}
+	if p != nil {
+		if p.Outcome != candidate.Outcome || p.PodUID != candidate.PodUID {
+			return false, fmt.Errorf("%w: process identity", errBackupSourceChanged)
+		}
+		return true, nil
+	}
+	encoded, err := json.Marshal(candidate)
+	if err != nil {
+		return false, err
+	}
+	patchBackupAnnotations(b, map[string]string{backupPreparationAnnotation: string(encoded)})
+	return true, r.Update(ctx, b)
+}
+
+// Current-copy workers are readers and are already protected by the owner;
+// ignore only those, so requeues can observe their completion. Any other PVC
+// user or unscheduled writer Job prevents stopped-source admission.
+func (r *BackupReconciler) backupSourceDrained(ctx context.Context, b *gameplanev1alpha1.Backup, gs *gameplanev1alpha1.GameServer, ss *appsv1.StatefulSet) (bool, error) {
+	var jobs batchv1.JobList
+	if err := r.backupAPIReader().List(ctx, &jobs, client.InNamespace(gs.Namespace)); err != nil {
+		return false, err
+	}
+	copyJobs := map[types.UID]bool{}
+	for i := range jobs.Items {
+		job := &jobs.Items[i]
+		if job.Name == b.Name && metav1.IsControlledBy(job, b) {
+			copyJobs[job.UID] = true
+			continue
+		}
+		if ((job.Status.Succeeded == 0 && !jobPermanentlyFailed(job)) || job.Status.Active > 0 || !job.DeletionTimestamp.IsZero()) && podMountsRestoreTarget(&corev1.Pod{Spec: job.Spec.Template.Spec}, gs.Name) {
+			return false, nil
+		}
+	}
+	var pods corev1.PodList
+	if err := r.backupAPIReader().List(ctx, &pods, client.InNamespace(gs.Namespace)); err != nil {
+		return false, err
+	}
+	for i := range pods.Items {
+		pod := &pods.Items[i]
+		if !liveRestorePod(pod) {
+			continue
+		}
+		owner := metav1.GetControllerOf(pod)
+		if owner != nil && owner.Kind == "Job" && copyJobs[owner.UID] {
+			continue
+		}
+		if metav1.IsControlledBy(pod, ss) || metav1.IsControlledBy(pod, gs) || pod.Name == gs.Name+"-0" || podMountsRestoreTarget(pod, gs.Name) {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+// An absent/replaced original pod cannot retain save-off state. A timeout,
+// unready pod, or terminating pod with the same UID is never proof of exit.
+// Empty UID is legacy state: only absence or an observed terminal pod suffices.
+func backupQuiesceProcessGone(ctx context.Context, reader client.Reader, namespace, server string, uid types.UID) (bool, error) {
+	var pod corev1.Pod
+	err := reader.Get(ctx, client.ObjectKey{Namespace: namespace, Name: server + "-0"}, &pod)
+	if apierrors.IsNotFound(err) {
+		return true, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	if uid != "" && pod.UID != uid {
+		return true, nil
+	}
+	return pod.DeletionTimestamp.IsZero() && (pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodFailed), nil
+}
diff --git a/operator/internal/controller/backup_stopped_test.go b/operator/internal/controller/backup_stopped_test.go
new file mode 100644
index 000000000..3d0f30f39
--- /dev/null
+++ b/operator/internal/controller/backup_stopped_test.go
@@ -0,0 +1,382 @@
+package controller
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1"
+	"github.com/GameplanePanel/gameplane/operator/internal/agent"
+)
+
+func stoppedBackupFixture(t *testing.T, suspended bool) (*BackupReconciler, *guardQuiescer) {
+	t.Helper()
+	r, q := guardFixture(t, guardBackup("first"), guardBackup("second"))
+	ctx := t.Context()
+	var gs gameplanev1alpha1.GameServer
+	key := client.ObjectKey{Namespace: "ns", Name: "server"}
+	if err := r.Get(ctx, key, &gs); err != nil {
+		t.Fatal(err)
+	}
+	gs.Spec.Suspend = suspended
+	if !suspended {
+		gs.Annotations = map[string]string{IdleAsleepSinceAnnotation: "2026-10-21T00:00:00Z"}
+	}
+	if err := r.Update(ctx, &gs); err != nil {
+		t.Fatal(err)
+	}
+	var ss appsv1.StatefulSet
+	if err := r.Get(ctx, key, &ss); err != nil {
+		t.Fatal(err)
+	}
+	zero := int32(0)
+	ss.Spec.Replicas = &zero
+	if err := r.Update(ctx, &ss); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Delete(ctx, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "server-0", Namespace: "ns"}}); err != nil {
+		t.Fatal(err)
+	}
+	return r, q
+}
+
+// A regression to unconditional agent quiesce would fail the first admission;
+// releasing before worker drain would let the second Job appear prematurely.
+func TestBackupStoppedCopiesWithoutWakingAndReleasesForNextCopy(t *testing.T) {
+	for _, suspended := range []bool{false, true} {
+		t.Run(map[bool]string{false: "idle", true: "suspended"}[suspended], func(t *testing.T) {
+			r, q := stoppedBackupFixture(t, suspended)
+			q.quiesceErr = errors.New("absent agent")
+			guardPass(t, r, "first")
+			first := guardGetBackup(t, r, "first")
+			var prep backupPreparation
+			if err := json.Unmarshal([]byte(first.Annotations[backupPreparationAnnotation]), &prep); err != nil {
+				t.Fatal(err)
+			}
+			if prep.Outcome != "already-stopped" || prep.TargetUID != "server-uid" || prep.WorkloadUID != "workload-uid" || prep.PodUID != "" {
+				t.Fatalf("wrong source proof: %+v", prep)
+			}
+			var job batchv1.Job
+			if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "first"}, &job); err != nil {
+				t.Fatal(err)
+			}
+			// A restart and concurrent wake request must preserve the zero-replica fence.
+			r = &BackupReconciler{Client: r.Client, APIReader: r.Client, Scheme: r.Scheme, AgentClient: q, LogReader: r.LogReader}
+			var gs gameplanev1alpha1.GameServer
+			if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "server"}, &gs); err != nil {
+				t.Fatal(err)
+			}
+			if gs.Annotations == nil {
+				gs.Annotations = map[string]string{}
+			}
+			gs.Annotations[IdleWakeRequestedAnnotation] = "wake"
+			if err := r.Update(t.Context(), &gs); err != nil {
+				t.Fatal(err)
+			}
+			gr := &GameServerReconciler{Client: r.Client, APIReader: r.Client, Scheme: r.Scheme}
+			if err := gr.reconcileStatefulSet(t.Context(), &gs, singlePortTemplate(true, "minecraft"), nil, nil, 1); err != nil {
+				t.Fatal(err)
+			}
+			var ss appsv1.StatefulSet
+			if err := r.Get(t.Context(), client.ObjectKeyFromObject(&gs), &ss); err != nil {
+				t.Fatal(err)
+			}
+			if ss.Spec.Replicas == nil || *ss.Spec.Replicas != 0 {
+				t.Fatal("wake crossed the backup fence")
+			}
+			guardPass(t, r, "second")
+			var second batchv1.Job
+			if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "second"}, &second); !apierrors.IsNotFound(err) {
+				t.Fatalf("second copy admitted before drain: %v", err)
+			}
+			job.Status.Succeeded = 1
+			if err := r.Status().Update(t.Context(), &job); err != nil {
+				t.Fatal(err)
+			}
+			guardPass(t, r, "first")
+			if got := guardGetBackup(t, r, "first"); got.Status.Phase != gameplanev1alpha1.BackupPhaseSucceeded || got.Status.SnapshotID == "" {
+				t.Fatalf("not restorable: %+v", got.Status)
+			}
+			guardPass(t, r, "second")
+			if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "second"}, &second); err != nil {
+				t.Fatal(err)
+			}
+			if q.quiesced != 0 || q.unquiesced != 0 {
+				t.Fatalf("stopped copy contacted absent agent: %+v", q)
+			}
+			if err := r.Get(t.Context(), client.ObjectKeyFromObject(&gs), &gs); err != nil {
+				t.Fatal(err)
+			}
+			if gs.Spec.Suspend != suspended || gs.Annotations[IdleWakeRequestedAnnotation] != "wake" || (!suspended && gs.Annotations[IdleAsleepSinceAnnotation] != "2026-10-21T00:00:00Z") {
+				t.Fatal("copy changed power intent")
+			}
+		})
+	}
+}
+
+func TestBackupStoppedRequiresLiveDrainProof(t *testing.T) {
+	for _, tc := range []string{"desired-one", "observed-replicas", "stale-generation", "terminating-game", "foreign-writer", "pending-writer-job", "read-error"} {
+		t.Run(tc, func(t *testing.T) {
+			r, q := stoppedBackupFixture(t, true)
+			base := r.Client
+			r.APIReader = interceptor.NewClient(base.(client.WithWatch), interceptor.Funcs{
+				Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+					if err := c.Get(ctx, key, obj, opts...); err != nil {
+						return err
+					}
+					if ss, ok := obj.(*appsv1.StatefulSet); ok {
+						switch tc {
+						case "desired-one":
+							one := int32(1)
+							ss.Spec.Replicas = &one
+						case "observed-replicas":
+							ss.Status.Replicas = 1
+						case "stale-generation":
+							ss.Generation = 2
+							ss.Status.ObservedGeneration = 1
+						}
+					}
+					return nil
+				},
+				List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
+					if err := c.List(ctx, list, opts...); err != nil {
+						return err
+					}
+					if pods, ok := list.(*corev1.PodList); ok {
+						if tc == "read-error" {
+							return errors.New("live list failed")
+						}
+						if tc == "terminating-game" {
+							now := metav1.Now()
+							pods.Items = append(pods.Items, corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "server-0", Namespace: "ns", DeletionTimestamp: &now}, Status: corev1.PodStatus{Phase: corev1.PodSucceeded}})
+						}
+						if tc == "foreign-writer" {
+							pods.Items = append(pods.Items, corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "writer", Namespace: "ns"}, Spec: corev1.PodSpec{Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "server-data"}}}}}})
+						}
+					}
+					if jobs, ok := list.(*batchv1.JobList); ok && tc == "pending-writer-job" {
+						jobs.Items = append(jobs.Items, batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "writer", Namespace: "ns"}, Spec: batchv1.JobSpec{Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "server-data"}}}}}}}})
+					}
+					return nil
+				},
+			})
+			_, _ = r.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKey{Namespace: "ns", Name: "first"}})
+			var job batchv1.Job
+			if err := base.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "first"}, &job); !apierrors.IsNotFound(err) {
+				t.Fatalf("copy admitted without drain proof: %v", err)
+			}
+			if q.quiesced != 0 {
+				t.Fatal("no proven live process to quiesce")
+			}
+		})
+	}
+}
+
+func TestBackupLiveTimeoutRetainsOwnerUntilOriginalPodGone(t *testing.T) {
+	r, q := guardFixture(t, guardBackup("first"), guardBackup("second"))
+	q.quiesceErr = errors.New("live agent unreachable")
+	q.unquiesceErr = errors.New("live agent still unreachable")
+	_, _ = r.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKey{Namespace: "ns", Name: "first"}})
+	first := guardGetBackup(t, r, "first")
+	if first.Status.Phase != gameplanev1alpha1.BackupPhaseFailed {
+		t.Fatalf("timeout did not fail closed: %+v", first.Status)
+	}
+	var job batchv1.Job
+	if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "first"}, &job); !apierrors.IsNotFound(err) {
+		t.Fatal("live timeout created copy")
+	}
+	var gs gameplanev1alpha1.GameServer
+	if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "server"}, &gs); err != nil {
+		t.Fatal(err)
+	}
+	op, _, err := lifecycleOwner(&gs)
+	if err != nil || op.QuiescePodUID != "game-pod-uid" || !op.NeedsUnquiesce {
+		t.Fatalf("cleanup identity not durable: %+v %v", op, err)
+	}
+	guardPass(t, r, "first")
+	guardPass(t, r, "second")
+	if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "second"}, &job); !apierrors.IsNotFound(err) {
+		t.Fatal("timeout released the live process owner")
+	}
+	cleanupAttempts := q.unquiesced
+	// The replacement must not receive its predecessor's delayed save-on.
+	if err := r.Delete(t.Context(), &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "server-0", Namespace: "ns"}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Create(t.Context(), &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "server-0", Namespace: "ns", UID: "replacement"}, Status: corev1.PodStatus{Phase: corev1.PodRunning}}); err != nil {
+		t.Fatal(err)
+	}
+	guardPass(t, r, "first")
+	if q.unquiesced != cleanupAttempts {
+		t.Fatal("replacement inherited original process cleanup")
+	}
+	if err := r.Get(t.Context(), client.ObjectKeyFromObject(&gs), &gs); err != nil {
+		t.Fatal(err)
+	}
+	if gs.Annotations[lifecycleOperationAnnotation] != "" {
+		t.Fatal("absent original process kept owner forever")
+	}
+}
+
+func TestBackupOldAgentFailsBeforeCleanupIntent(t *testing.T) {
+	r, q := guardFixture(t, guardBackup("first"))
+	q.checkErr = errors.New("identity endpoint: status 404")
+	_, _ = r.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKey{Namespace: "ns", Name: "first"}})
+	b := guardGetBackup(t, r, "first")
+	if b.Status.Phase != gameplanev1alpha1.BackupPhaseFailed || b.Annotations[annoQuiesceAttempted] != "" || q.quiesced != 0 {
+		t.Fatalf("unsupported identity created cleanup debt: backup=%+v agent=%+v", b, q)
+	}
+	guardPass(t, r, "first")
+	var gs gameplanev1alpha1.GameServer
+	if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "server"}, &gs); err != nil {
+		t.Fatal(err)
+	}
+	if gs.Annotations[lifecycleOperationAnnotation] != "" {
+		t.Fatal("old agent failure blocked the workload upgrade")
+	}
+}
+
+func TestBackupDisabledAgentPreservesRawCopyWithoutClaimingQuiesce(t *testing.T) {
+	r, _ := guardFixture(t, guardBackup("first"))
+	r.AgentClient = &agent.Client{Disabled: true}
+	guardPass(t, r, "first")
+	b := guardGetBackup(t, r, "first")
+	if b.Annotations[backupPreparationAnnotation] != "" || b.Annotations[annoQuiesceAttempted] != "" {
+		t.Fatal("disabled agent claimed preparation or quiesce")
+	}
+	var job batchv1.Job
+	if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "first"}, &job); err != nil {
+		t.Fatalf("disabled client prevented raw copy: %v", err)
+	}
+	if err := callBackupQuiesce(t.Context(), r.AgentClient, "ns", "server", "", true); err == nil {
+		t.Fatal("disabled client cleared an existing cleanup obligation")
+	}
+}
+
+func TestBackupDisabledAgentCannotBypassPendingLiveQuiesce(t *testing.T) {
+	r, _ := guardFixture(t, guardBackup("first"))
+	b := guardGetBackup(t, r, "first")
+	if ok, err := r.claimBackupTarget(t.Context(), b); err != nil || !ok {
+		t.Fatalf("claim: %t %v", ok, err)
+	}
+	gs, err := r.ownedBackupTarget(t.Context(), b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ready, err := r.prepareBackupSource(t.Context(), b, gs); err != nil || !ready {
+		t.Fatalf("prepare: %t %v", ready, err)
+	}
+	patchBackupAnnotations(b, map[string]string{annoQuiesceAttempted: "pending"})
+	if err := r.Update(t.Context(), b); err != nil {
+		t.Fatal(err)
+	}
+	r.AgentClient = &agent.Client{Disabled: true}
+	if err := r.maybeQuiesce(t.Context(), b); err == nil || b.Status.Phase != gameplanev1alpha1.BackupPhaseFailed {
+		t.Fatalf("disabled client bypassed pending quiesce: %v %+v", err, b.Status)
+	}
+}
+
+func TestBackupSourceReplacementFailsBeforeCopyOrSuccess(t *testing.T) {
+	for _, duringRPC := range []bool{true, false} {
+		t.Run(map[bool]string{true: "during-quiesce", false: "during-copy"}[duringRPC], func(t *testing.T) {
+			r, q := guardFixture(t, guardBackup("first"))
+			replace := func() {
+				var old corev1.Pod
+				if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "server-0"}, &old); err != nil {
+					t.Fatal(err)
+				}
+				if err := r.Delete(t.Context(), &old); err != nil {
+					t.Fatal(err)
+				}
+				old.UID, old.ResourceVersion = "replacement", ""
+				if err := r.Create(t.Context(), &old); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if duringRPC {
+				q.afterQuiesce = replace
+			} else {
+				guardPass(t, r, "first")
+				var job batchv1.Job
+				if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "first"}, &job); err != nil {
+					t.Fatal(err)
+				}
+				job.Status.Succeeded = 1
+				if err := r.Status().Update(t.Context(), &job); err != nil {
+					t.Fatal(err)
+				}
+				replace()
+			}
+			guardPass(t, r, "first")
+			if got := guardGetBackup(t, r, "first"); got.Status.Phase != gameplanev1alpha1.BackupPhaseFailed || got.Status.SnapshotID != "" {
+				t.Fatalf("replacement allowed a restorable copy: %+v", got.Status)
+			}
+			if duringRPC {
+				var job batchv1.Job
+				if err := r.Get(t.Context(), client.ObjectKey{Namespace: "ns", Name: "first"}, &job); !apierrors.IsNotFound(err) {
+					t.Fatal("replacement during RPC created a copy")
+				}
+			}
+			if q.unquiesced != 0 {
+				t.Fatal("replacement received the old process cleanup")
+			}
+		})
+	}
+}
+
+func TestBackupCleanupRequiresOriginalProcessExit(t *testing.T) {
+	for _, state := range []string{"absent", "replacement", "terminal", "terminating", "unready", "legacy-running", "read-error"} {
+		t.Run(state, func(t *testing.T) {
+			r, _ := guardFixture(t)
+			uid := types.UID("game-pod-uid")
+			var pod corev1.Pod
+			key := client.ObjectKey{Namespace: "ns", Name: "server-0"}
+			if err := r.Get(t.Context(), key, &pod); err != nil {
+				t.Fatal(err)
+			}
+			reader := interceptor.NewClient(r.Client.(client.WithWatch), interceptor.Funcs{Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+				if state == "read-error" {
+					return errors.New("apiserver unavailable")
+				}
+				if state == "absent" {
+					return apierrors.NewNotFound(corev1.Resource("pods"), key.Name)
+				}
+				if err := c.Get(ctx, key, obj, opts...); err != nil {
+					return err
+				}
+				p := obj.(*corev1.Pod)
+				switch state {
+				case "replacement":
+					p.UID = "successor"
+				case "terminal":
+					p.Status.Phase = corev1.PodFailed
+				case "terminating":
+					now := metav1.Now()
+					p.DeletionTimestamp = &now
+				case "unready":
+					p.Status.Phase = corev1.PodPending
+				}
+				return nil
+			}})
+			if state == "legacy-running" {
+				uid = ""
+			}
+			gone, err := backupQuiesceProcessGone(t.Context(), reader, "ns", "server", uid)
+			wantGone := state == "absent" || state == "replacement" || state == "terminal"
+			if gone != wantGone || (err != nil) != (state == "read-error") {
+				t.Fatalf("exit proof = %t, %v; want gone %t", gone, err, wantGone)
+			}
+		})
+	}
+}
diff --git a/operator/internal/controller/gameserver_controller.go b/operator/internal/controller/gameserver_controller.go
index b072d5946..1c6982277 100644
--- a/operator/internal/controller/gameserver_controller.go
+++ b/operator/internal/controller/gameserver_controller.go
@@ -2387,6 +2387,7 @@ func buildAgentContainer(
 	env := []corev1.EnvVar{
 		{Name: "GAMEPLANE_SERVER_NAME", Value: gs.Name},
 		{Name: "GAMEPLANE_SERVER_UID", Value: string(gs.UID)},
+		{Name: "GAMEPLANE_POD_UID", ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.uid"}}},
 		{Name: "GAMEPLANE_TEMPLATE", Value: tmpl.Name},
 		{Name: "GAMEPLANE_GAME", Value: tmpl.Spec.Game},
 		// Games without RCON (consoleMode pty/none) must not have the
diff --git a/operator/internal/controller/helpers_envtest_test.go b/operator/internal/controller/helpers_envtest_test.go
index 927d58443..eaf4cb6c5 100644
--- a/operator/internal/controller/helpers_envtest_test.go
+++ b/operator/internal/controller/helpers_envtest_test.go
@@ -780,6 +780,14 @@ func (f *fakeAgent) Unquiesce(_ context.Context, _, _ string) error {
 	return f.unquiesceErr
 }
 
+func (f *fakeAgent) CheckPod(context.Context, string, string, string) error { return nil }
+func (f *fakeAgent) QuiescePod(ctx context.Context, namespace, server, _ string) error {
+	return f.Quiesce(ctx, namespace, server)
+}
+func (f *fakeAgent) UnquiescePod(ctx context.Context, namespace, server, _ string) error {
+	return f.Unquiesce(ctx, namespace, server)
+}
+
 func (f *fakeAgent) quiesceCount() int {
 	f.mu.Lock()
 	defer f.mu.Unlock()
diff --git a/operator/internal/controller/lifecycle_operation.go b/operator/internal/controller/lifecycle_operation.go
index fc769d5e7..fc1d626fb 100644
--- a/operator/internal/controller/lifecycle_operation.go
+++ b/operator/internal/controller/lifecycle_operation.go
@@ -31,6 +31,7 @@ type lifecycleIdentity struct {
 	Generation     int64     `json:"generation"`
 	Token          string    `json:"token,omitempty"`
 	NeedsUnquiesce bool      `json:"needsUnquiesce,omitempty"`
+	QuiescePodUID  types.UID `json:"quiescePodUID,omitempty"`
 	// InitialStartup permits only the first workload of a CSI restore's new
 	// target (including WaitForFirstConsumer PVCs), while other operations wait.
 	InitialStartup bool `json:"initialStartup,omitempty"`
@@ -295,15 +296,21 @@ func recoverLifecycleOperation(ctx context.Context, c client.Client, reader clie
 			return false, err
 		}
 	}
-	needsUnquiesce := false
 	for _, predecessor := range op.Recovery {
-		needsUnquiesce = needsUnquiesce || predecessor.NeedsUnquiesce
-	}
-	if needsUnquiesce {
-		if quiescer == nil {
+		if !predecessor.NeedsUnquiesce {
+			continue
+		}
+		gone, err := backupQuiesceProcessGone(ctx, reader, gs.Namespace, gs.Name, predecessor.QuiescePodUID)
+		if err != nil {
+			return false, err
+		}
+		if gone {
+			continue
+		}
+		if !agentQuiesceAvailable(quiescer) {
 			return false, fmt.Errorf("cannot recover orphaned quiesce without an agent client")
 		}
-		if err := quiescer.Unquiesce(ctx, gs.Namespace, gs.Name); err != nil {
+		if err := callBackupQuiesce(ctx, quiescer, gs.Namespace, gs.Name, predecessor.QuiescePodUID, true); err != nil {
 			return false, fmt.Errorf("recover orphaned lifecycle quiesce: %w", err)
 		}
 	}
@@ -355,10 +362,23 @@ func (r *BackupReconciler) setLifecycleUnquiesceIntent(ctx context.Context, b *g
 	if err != nil {
 		return err
 	}
-	if op.NeedsUnquiesce == needed {
+	var podUID types.UID
+	if needed {
+		preparation, err := parseBackupPreparation(b)
+		if err != nil {
+			return err
+		}
+		if preparation != nil {
+			podUID = preparation.PodUID
+		}
+	}
+	if op.NeedsUnquiesce == needed && (!needed || op.QuiescePodUID == podUID) {
 		return nil
 	}
 	op.NeedsUnquiesce = needed
+	if needed {
+		op.QuiescePodUID = podUID
+	}
 	gs.Annotations[lifecycleOperationAnnotation] = lifecycleOperationJSON(op)
 	return r.Update(ctx, gs)
 }
diff --git a/operator/specs.md b/operator/specs.md
index dc327718e..aeb79ff64 100644
--- a/operator/specs.md
+++ b/operator/specs.md
@@ -249,6 +249,10 @@ Primary reconcilers register with the manager in `cmd/main.go` and handle CRD li
 - **Responsibility:** Drive Backup to completion: coordinate quiesce (if agent available), spawn restic Job, track snapshot ID.
 - **Key functions:**
   - Quiesce orchestration: call agent Quiesce before Job, Unquiesce after completion.
+  - With quiesce and an agent client enabled, preparation is recorded once in `backup.gameplane.local/preparation`, bound to the target UID and StatefulSet UID/generation. `already-stopped` requires uncached proof of desired and observed zero replicas, caught-up observed generation, and no live game/PVC user pods or pending writer Jobs. Current copy workers are excluded by Job ownership. This path sends no agent calls and preserves suspend/idle/wake intent under the shared workload fence. It applies equally to restic and CSI copies.
+  - Live preparation also pins the game Pod UID. The operator checks the agent's read-only `/v1/pods/{uid}/identity` endpoint before recording the first cleanup intent, then uses pod-bound quiesce/unquiesce routes; the agent rejects a different UID before RCON. A missing endpoint or identity mismatch fails closed without a legacy-route fallback. Ambiguous quiesce errors retain cleanup intent and the original Pod UID on both the Backup and shared owner. Only confirmed save-on or uncached evidence that the original pod is absent/replaced/terminal permits cleanup; an HTTP timeout, unready pod, or terminating pod is insufficient. Legacy attempts without a UID retire only on absence/terminal proof and never adopt a new UID. `Unquiesced=True, Reason=SourceProcessGone` distinguishes process exit from an actual save-on RPC.
+  - Source proof is checked before and after quiesce and on each copy reconcile. Observed workload/process replacement fails the Backup before it can report a restorable success. Owner recovery retains the pod UID even if the Backup is force-deleted, drains predecessor workers, and never sends its cleanup to a successor pod.
+  - Upgrade the operator and agent image together and allow existing GameServer pods to roll out the new agent and `GAMEPLANE_POD_UID` DownwardAPI env. Older agents fail the identity preflight before cleanup debt, so the workload can continue its upgrade after terminal cleanup releases the owner. Retry failed copies after rollout; preflight never bypasses protection. This does not automatically stop a running game.
   - Every copy, including `quiesce=false` and CSI snapshots, acquires the shared lifecycle owner before starting. The Backup pins its target in `backup.gameplane.local/target`; `backup.gameplane.local/owner` remains a compatibility marker. The shared owner is described below. Quiesce cleanup intent is recorded on both the Backup and the surviving GameServer before calling the agent, including partial failures. Cleanup waits for live copy workers; deletion cancels the copy Job with foreground propagation first. Every Backup carries the cleanup finalizer. Pre-upgrade quiesced workers are observed through completion even when overlapping legacy copies cannot acquire the new exclusive owner.
   - Orphan recovery preserves predecessor operation identities in the shared owner's `recovery` array, atomically with ownership transfer. Existing `backup.gameplane.local/recovering-owners` lineage is imported during migration. Deleting a recovering owner cannot hide an earlier worker. Normal deletion, orphan recovery, and legacy cleanup also wait for pending owned CSI captures to report ready or an explicit error; VolumeSnapshots are reclaimed by owner-reference GC after cleanup.
   - VolumeSnapshot API availability is checked before quiescing. Snapshot creation errors fail the Backup and run unquiesce cleanup immediately instead of retrying creation indefinitely with saving disabled. API/creation failures persist `backup.gameplane.local/snapshot-aborted` to allow cleanup without waiting for a failed/nonrestorable capture. Failed unquiesce remains visible and retried. A transport failure can leave a CSI snapshot object despite a Failed Backup; this does not make that Backup restorable.
@@ -517,7 +521,7 @@ Forgetting codegen leaves the YAML out of sync with types — CI's `make manifes
 
 6. **Module sources are immutable once pulled.** Digest pinning (spec.digest) defeats tag moves; version pinning (spec.version) tracks a specific semver. Floating (unset version/digest) tracks latest and re-pulls on ModuleSource refresh.
 
-7. **Backup quiesce is best-effort.** If agent unavailable, backup proceeds raw (no pause). If quiesce unsupported (agent returns ErrUnsupported), backup continues degraded (success-with-note).
+7. **Backup quiesce fails closed on live transport errors.** With no configured agent client or `quiesce=false`, a copy proceeds raw. With a client enabled, a drained stopped source skips RPC, but an unreachable live agent prevents the copy. A game that explicitly reports quiesce unsupported can continue without claiming it was quiesced.
 
 8. **Remote cluster health checks are non-blocking.** A Cluster with health Unhealthy does not prevent GameServer creation on the local cluster; it surfaces the issue so operators can intervene.
 
diff --git a/test/e2e/backup_stopped_e2e_test.go b/test/e2e/backup_stopped_e2e_test.go
new file mode 100644
index 000000000..e02258db2
--- /dev/null
+++ b/test/e2e/backup_stopped_e2e_test.go
@@ -0,0 +1,115 @@
+//go:build e2e
+
+package e2e
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"regexp"
+	"testing"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// Exercise real PVC data and restic for two consecutive protected copies while
+// the agent is absent. This test needs no API login and runs in the operator bucket.
+func TestBackup_StoppedCopiesRemainRestorable(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	suffix := time.Now().UnixNano()
+	ns := "gameplane-games"
+	gs := fmt.Sprintf("e2e-stopped-backup-%d", suffix)
+	tmpl := fmt.Sprintf("e2e-stopped-backup-template-%d", suffix)
+	ensureResticRepo(t)
+	applyBusyboxTemplate(t, tmpl)
+	applyBusyboxGameServer(t, ns, gs, tmpl)
+	waitPVCBound(t, ns, gs+"-data", 90*time.Second)
+	envInstance.Eventually(t, 2*time.Minute, func() (bool, string) {
+		pod, err := envInstance.K8s.CoreV1().Pods(ns).Get(ctx, gs+"-0", metav1.GetOptions{})
+		if err != nil {
+			return false, err.Error()
+		}
+		for _, status := range pod.Status.ContainerStatuses {
+			if status.Name == "game" && status.Ready {
+				return true, ""
+			}
+		}
+		return false, "game container not ready"
+	})
+	if out, err := envInstance.KubectlExec(t, ns, "pod/"+gs+"-0", "sh", "-c", "echo stopped-backup-marker > /data/marker.txt && sync"); err != nil {
+		t.Fatalf("write marker: %v\n%s", err, out)
+	}
+	if _, err := envInstance.Dyn.Resource(gameServerGVR).Namespace(ns).Patch(ctx, gs, types.MergePatchType, []byte(`{"spec":{"suspend":true}}`), metav1.PatchOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	envInstance.Eventually(t, 3*time.Minute, func() (bool, string) {
+		ss, err := envInstance.K8s.AppsV1().StatefulSets(ns).Get(ctx, gs, metav1.GetOptions{})
+		if err != nil {
+			return false, err.Error()
+		}
+		if ss.Spec.Replicas == nil || *ss.Spec.Replicas != 0 || ss.Status.Replicas != 0 || ss.Status.ObservedGeneration < ss.Generation {
+			return false, "workload not drained"
+		}
+		_, err = envInstance.K8s.CoreV1().Pods(ns).Get(ctx, gs+"-0", metav1.GetOptions{})
+		return apierrors.IsNotFound(err), fmt.Sprintf("pod absence: %v", err)
+	})
+	for i := 1; i <= 2; i++ {
+		name := fmt.Sprintf("e2e-stopped-copy-%d-%d", suffix, i)
+		backup := &unstructured.Unstructured{Object: map[string]any{
+			"apiVersion": "gameplane.local/v1alpha1", "kind": "Backup",
+			"metadata": map[string]any{"name": name, "namespace": ns},
+			"spec":     map[string]any{"serverRef": map[string]any{"name": gs}, "repoRef": map[string]any{"name": "e2e-restic-creds", "key": "repo"}, "strategy": "restic-snapshot", "quiesce": true},
+		}}
+		if _, err := envInstance.Dyn.Resource(backupGVR).Namespace(ns).Create(ctx, backup, metav1.CreateOptions{}); err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { deleteBackupCR(t, ns, name) })
+		var snapshotID string
+		envInstance.Eventually(t, 5*time.Minute, func() (bool, string) {
+			// Check throughout the copy, not only after its terminal result.
+			_, err := envInstance.K8s.CoreV1().Pods(ns).Get(ctx, gs+"-0", metav1.GetOptions{})
+			if !apierrors.IsNotFound(err) {
+				t.Fatalf("backup woke game or absence could not be confirmed: %v", err)
+			}
+			got, err := envInstance.Dyn.Resource(backupGVR).Namespace(ns).Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				return false, err.Error()
+			}
+			phase, _, _ := unstructured.NestedString(got.Object, "status", "phase")
+			if phase == "Failed" {
+				t.Fatalf("stopped backup failed: %+v", got.Object["status"])
+			}
+			if phase != "Succeeded" {
+				return false, "phase=" + phase
+			}
+			var preparation struct {
+				Outcome string `json:"outcome"`
+			}
+			if err := json.Unmarshal([]byte(got.GetAnnotations()["backup.gameplane.local/preparation"]), &preparation); err != nil || preparation.Outcome != "already-stopped" {
+				t.Fatalf("missing stopped-source proof: %+v, %v", preparation, err)
+			}
+			if got.GetAnnotations()["backup.gameplane.local/quiesce-attempted"] != "" {
+				t.Fatal("stopped source contacted absent agent")
+			}
+			snapshotID, _, _ = unstructured.NestedString(got.Object, "status", "snapshotID")
+			return snapshotID != "", "snapshot ID pending"
+		})
+		if !regexp.MustCompile(`^[0-9a-f]{8,64}$`).MatchString(snapshotID) {
+			t.Fatalf("unexpected snapshot ID %q", snapshotID)
+		}
+		runResticJob(t, "restic restore "+snapshotID+" --target /tmp/stopped-restore --retry-lock 5m && test \"$(cat /tmp/stopped-restore/data/marker.txt)\" = stopped-backup-marker")
+	}
+	got, err := envInstance.Dyn.Resource(gameServerGVR).Namespace(ns).Get(ctx, gs, metav1.GetOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	suspended, _, _ := unstructured.NestedBool(got.Object, "spec", "suspend")
+	if !suspended {
+		t.Fatal("backup changed manual suspend intent")
+	}
+}
diff --git a/test/e2e/buckets.sh b/test/e2e/buckets.sh
index b803e2f42..6686774d7 100755
--- a/test/e2e/buckets.sh
+++ b/test/e2e/buckets.sh
@@ -84,6 +84,7 @@ TestModuleSourceUpload
 TestModule_ScaffoldAndPackage
 TestModule_SteamcmdScaffoldKeepsNonRootSecurity
 TestBackup_OperatorMaterializesJob
+TestBackup_StoppedCopiesRemainRestorable
 TestBackup_FailsOnMissingPVC
 TestBackup_FailsOnBadCredentials
 TestBackup_DeleteForgetsSnapshot